uranusjr commented on code in PR #74701:
URL: https://github.com/apache/airflow/pull/74701#discussion_r4335727776
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -161,853 +108,675 @@ Configure the build in ``build.gradle``:
}
airflowBundle {
- mainClass = "com.mycompany.airflow.sales.Main"
+ mainClass = "com.mycompany.airflow.hello.Main"
}
-The ``annotationProcessor`` dependency is required for this annotation-based
example. Omit it when using the
-interface-based API. If the project does not have the Gradle Wrapper yet,
generate it from the project root
-with ``gradle wrapper``.
-
-Implement the Java tasks
-~~~~~~~~~~~~~~~~~~~~~~~~
-
-Save the task implementations as
-``src/main/java/com/mycompany/airflow/sales/SalesPipeline.java``:
+Declare the Dag in ``HelloDag.java``:
.. code-block:: java
- package com.mycompany.airflow.sales;
+ package com.mycompany.airflow.hello;
- import org.apache.airflow.sdk.Airflow;
+ import org.apache.airflow.sdk.*;
- public class SalesPipeline {
+ @Airflow.Dag(id = "java_hello", schedule = "@daily")
+ public class HelloDag {
- @Airflow.TaskHandler(dag = "sales_pipeline", task = "extract")
+ @Airflow.Task(queue = "java")
public long extract() {
- return 3;
+ return 42L;
}
- @Airflow.TaskHandler(dag = "sales_pipeline", task = "transform")
- public long transform(long recordCount) {
- return recordCount * 2;
+ @Airflow.Task(queue = "java")
+ public void report(long rows) {
+ System.out.println("extracted " + rows + " rows");
}
- }
-
-.. note::
-
- The graph is declared once, in the Python Dag file: ``transform(extract())``
feeds the upstream's
- return value into the downstream's parameter by calling tasks like
functions. The supervisor sends
- the resulting *argument bindings* to the Java runtime, and each Java data
parameter receives
- whatever the Python call site bound at its position — an upstream task's
XCom or an inline
- literal. See :ref:`java-sdk/arg-binding`.
-Add the Java entry point
-~~~~~~~~~~~~~~~~~~~~~~~~
+ @Airflow.Deps
+ static class Wiring implements HelloDagDeps {
+ void depends() {
+ report(extract());
+ }
+ }
+ }
-Save the entry point as
``src/main/java/com/mycompany/airflow/sales/Main.java``:
+Serve it from ``Main.java``:
.. code-block:: java
- package com.mycompany.airflow.sales;
+ package com.mycompany.airflow.hello;
import org.apache.airflow.sdk.Bundle;
import org.apache.airflow.sdk.Server;
public class Main {
public static void main(String[] args) {
- Server.create(args).serve(new Bundle().register(SalesPipeline.class));
+ Server.create(args).serve(new Bundle().register(HelloDag.class));
}
}
-``register`` takes the class the annotations are on, so there is no second
name to keep in sync: the
-annotation processor generates the registrar it reads during compilation.
-
-Build and deploy
-~~~~~~~~~~~~~~~~
-
-Run the bundle task from the ``sales-pipeline-java/`` project root:
+Build the bundle from the project root:
.. code-block:: bash
./gradlew bundle
-The deployable JAR or JARs are now in ``build/bundle/``. Copy or mount that
entire directory into a path
-available on every worker that can consume the ``java`` queue. For example,
deploy it as
-``/opt/airflow/jars/sales-pipeline/``. Java source files do not need to be
deployed to Airflow.
+Copy ``build/bundle/`` into a Dag bundle, for example a ``java-hello/``
directory in the default ``dags-folder``
+bundle, which reads the ``[core] dags_folder`` directory (see
+:doc:`Dag bundles </administration-and-deployment/dag-bundles>`). Then send
the ``java`` queue to the Java
+coordinator in ``airflow.cfg``:
-For a local deployment where those paths are writable, the copy commands could
be:
+.. code-block:: ini
-.. code-block:: bash
+ [sdk]
+ coordinators = {"java": {"classpath":
"airflow.sdk.coordinators.java.JavaCoordinator"}}
+ queue_to_coordinator = {"java": "java"}
- mkdir -p /opt/airflow/jars/sales-pipeline
- cp build/bundle/*.jar /opt/airflow/jars/sales-pipeline/
+Airflow reads the bundle like any other Dag file: ``java_hello`` shows up in
the UI, runs on its schedule,
+and ``report`` receives the value ``extract`` returned.
-Deploy ``sales_pipeline.py`` separately through the deployment's normal Dag
delivery process. For example,
-that process might sync it to ``${AIRFLOW_HOME}/dags/`` or package it in a Dag
bundle; neither location is
-inside or relative to ``sales-pipeline-java/``.
+.. _java-sdk/dag-definition:
-Configure Airflow to register the JAR directory as a Dag bundle, point the
coordinator at it, and route the
-``java`` queue to the coordinator. Add the following sections to the file
selected by ``AIRFLOW_CONFIG`` (by
-default, ``${AIRFLOW_HOME}/airflow.cfg``), or set the equivalent
``AIRFLOW__*`` environment variables:
+Declaring a Dag in Java
+-----------------------
-.. code-block:: ini
+A Dag can be declared in two styles, and each section below shows both:
- [dag_processor]
- dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {
- "name": "java-task-handlers",
- "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
- "kwargs": {"path": "/opt/airflow/jars/sales-pipeline"}
- }
- ]
+* **Annotations.** ``@Airflow.Dag`` marks a class as a Dag, ``@Airflow.Task``
marks its task methods, and a
+ nested ``@Airflow.Deps`` class wires them together. The annotation processor
generates the code that
+ connects them, so ``javac`` type-checks the graph.
+* **Interface.** Each task is a class implementing ``Task``, and a ``DagDef``
object registers the tasks and
+ their dependencies. It needs no annotation processor.
- [sdk]
- coordinators = {
- "java": {
- "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
- "kwargs": {"task_handler_bundle_name": "java-task-handlers"}
- }
- }
- queue_to_coordinator = {"java": "java"}
+Tasks and their inputs
+~~~~~~~~~~~~~~~~~~~~~~
-``java`` is a user-chosen coordinator name, not a reserved value. The value
assigned to the queue in
-``queue_to_coordinator`` must match a key in ``coordinators``, and
``task_handler_bundle_name`` must match a
-Dag bundle name in ``dag_bundle_config_list``. See
:ref:`java-sdk/coordinator-config` for how JARs are
-located.
+With annotations, ``@Airflow.Task`` marks a method as a task, and the nested
``@Airflow.Deps`` class declares
+the graph by calling the task methods:
-Restart the affected Airflow components after changing this configuration. The
coordinator config, the JARs
-and a JRE must be available wherever tasks execute and on the Dag processor.
With ``CeleryExecutor``, tasks
-execute on the Celery workers; with ``LocalExecutor``, they run in
subprocesses on the scheduler's host. The
-Dag processor checks the stub tasks of ``sales_pipeline.py`` against the task
handlers the JARs register, so
-it runs them too. The API server does not need any of it. Register the Dag
bundle in
-``[dag_processor] dag_bundle_config_list`` on every component, like your other
Dag bundles: the worker and
-the Dag processor resolve ``task_handler_bundle_name`` through it, and
wherever the ``[sdk]`` config is read
-it is rejected if the name is missing there. The Dag processor still receives
``sales_pipeline.py`` through
-the separate Dag delivery process.
+.. code-block:: java
-A Dag processor with this ``[sdk]`` configuration also parses the executable
JARs of every Dag bundle,
-and needs a JRE to do so (see :ref:`java-sdk/native-dag-parsing`). Keep it
from parsing a bundle's JARs
-with that bundle's ``.airflowignore``, whose pattern syntax follows ``[core]
dag_ignore_file_syntax``.
+ @Airflow.Dag(id = "java_etl", schedule = "@daily")
+ public class EtlPipeline {
-With ``task_handler_bundle_name`` set to its own bundle, as above,
``java-task-handlers`` holds only the
-JARs the Python Dag's tasks run, so ignoring everything in it is safe:
+ @Airflow.Task(retries = 2)
+ public long extract() {
+ return 42L;
+ }
-.. code-block:: bash
+ @Airflow.Task
+ public long transform(long extracted, double factor) {
+ return (long) (extracted * factor);
+ }
- # [core] dag_ignore_file_syntax = glob (the default)
- echo '*' > /opt/airflow/jars/sales-pipeline/.airflowignore
+ @Airflow.Task
+ public void load(long transformed) {}
- # [core] dag_ignore_file_syntax = regexp; a bare * is not a valid pattern
and is dropped
- echo '.' > /opt/airflow/jars/sales-pipeline/.airflowignore
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ load(transform(extract(), lit(0.9)));
+ }
+ }
+ }
-With ``task_handler_bundle_name`` unset, the JAR sits in the same Dag bundle
as ``sales_pipeline.py``, so
-ignore only the JAR, not the whole bundle:
+The annotation processor generates the ``EtlPipelineDeps`` interface, with one
method per task. Passing the
+value one call returned into another call makes the second task wait for the
first and receive its value.
+``lit(...)`` passes a constant instead, which must be a string, number,
boolean, list or map.
-.. code-block:: bash
+* Every task method is called in the wiring class. A task the wiring leaves
out fails when the Dag is read.
+* A task method that takes arguments is called once; keep its result in a
local variable to use it twice.
+* Values move between tasks as XComs, so a task returns data JSON can hold
(see :ref:`java-sdk/types`).
- # [core] dag_ignore_file_syntax = glob (the default)
- echo '*.jar' >> /opt/airflow/dags/.airflowignore
+With the interface, each task is a class, and ``dag.task(taskId, TaskClass)``
registers it. A task reads the
+value of an upstream task itself, through ``client.getXCom``:
- # [core] dag_ignore_file_syntax = regexp
- echo '\.jar$' >> /opt/airflow/dags/.airflowignore
+.. code-block:: java
-After Airflow has parsed the Dag, trigger it from the UI or command line:
+ public class Extract implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ client.setXCom(42L);
+ }
+ }
-.. code-block:: bash
+ public class Load implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ var extracted = ((Number) client.getXCom("extract")).longValue();
+ }
+ }
- airflow dags trigger sales_pipeline
+ var dag = new DagDef("java_etl").config("schedule", "@daily");
+ var extract = dag.task("extract", Extract.class).config("retries", 2);
+ var load = dag.task("load", Load.class);
+ extract.before(load);
-The Java ``extract`` task returns ``3``, ``transform`` returns ``6`` through
XCom, and the Python ``load`` task
-logs ``Loaded: 6``.
+A task class must be ``public`` and concrete, have a public no-argument
constructor, and be ``static`` if it is
+nested in another class. The runtime creates a new instance for every task run.
-See :ref:`java-sdk/coordinator-config` for the full list of accepted
``kwargs``.
-For a larger example that exercises connections, variables, logging, and both
task APIs, see the
-`Java SDK example in the Airflow repository
<https://github.com/apache/airflow/tree/main/java-sdk/example>`__.
+Register either kind of Dag on the ``Bundle`` that ``main`` serves:
-Writing tasks
--------------
+.. code-block:: java
-The Java SDK offers two APIs for implementing tasks. Both produce the same
runtime behavior; the choice is a
-matter of style.
+ Server.create(args).serve(new
Bundle().register(EtlPipeline.class).register(dag));
-.. _java-sdk/annotation-api:
+Task ids
+~~~~~~~~
-Annotation-based API
-~~~~~~~~~~~~~~~~~~~~
+``@Airflow.Task`` takes the task id from the method name unless ``id`` is set,
and ``@Airflow.Dag`` takes the
+Dag id from the class name unless ``id`` is set. ``dag.If`` and ``dag.Switch``
take it from the class name unless
+you pass one.
-Annotate a plain Java class and let the SDK generate the boilerplate at
compile time.
+Dag and task options
+~~~~~~~~~~~~~~~~~~~~
-.. list-table::
- :header-rows: 1
- :widths: 30 70
-
- * - Annotation
- - Purpose
- * - ``@Airflow.TaskHandler(dag = "...", task = "...")``
- - Marks a method as the Java body of a task the Python Dag file declares
with ``@task.stub``.
- ``dag`` must match the ``dag_id`` and ``task`` the stub function name;
omitting ``task``
- derives it from the method name. There is no class-level annotation on
this surface — the
- Dag is the Python file's, so the handler names the pair it binds to.
- * - ``@Airflow.Dag(id = "...")``
- - Marks a class as a Dag that Java itself owns. Attributes
(``schedule``, ``description``,
- ``tags``, ``catchup``, …) are Airflow's own Dag settings; only
attributes written
- explicitly are applied. See :ref:`java-sdk/native-dags`.
- * - ``@Airflow.Task(id = "...")``
- - Marks a method as a task of a Java-owned Dag. If ``id`` is omitted the
method name is
- used. Further attributes (``retries``, ``queue``, ``retryDelay``, …)
are Airflow's own
- task settings; only attributes written explicitly are applied.
- * - ``@Airflow.Deps``
- - Marks the nested class that declares the task graph in Java,
TaskFlow-style. Required for
- a Dag that Java owns end to end. See :ref:`java-sdk/native-dags`.
- * - ``TaskInput`` / ``@ArgName("...")``
- - Marks a class as a task's input, so keyword arguments bind by name
instead of by position:
- each public field receives the argument whose name matches it, ignoring
case and
- underscores. ``@ArgName`` pins a name the match cannot reach, or
renames the argument
- deliberately. See :ref:`java-sdk/arg-binding`.
-
-Besides the annotations, a task method may declare a ``Client`` and a
``Context`` parameter in any
-position; the SDK injects both. Every other parameter is a *data parameter*
and receives an
-argument bound by the Python ``@task.stub`` call site.
-
-The annotation processor generates a ``<ClassName>Builder`` class that wires
up the task
-registry and resolves data parameters and XCom pushes automatically.
+The attributes of ``@Airflow.Dag`` and ``@Airflow.Task``, and the keys of
``DagDef.config`` and
+``TaskDef.config``, are Airflow's own Dag and task settings. Annotation
attributes are ``camelCase``
+(``retryDelay``), and ``config`` keys are spelled as in Python
(``"retry_delay"``). Only the settings you write
+are applied, so Airflow's defaults hold for the rest.
.. code-block:: java
+ @Airflow.Dag(id = "java_etl", schedule = "@daily", startDate =
"2026-01-01T00:00:00Z", catchup = false)
- public class MyDag {
+ @Airflow.Task(retries = 2, retryDelay = "PT5M", queue = "java")
- @Airflow.TaskHandler(dag = "my_dag", task = "fetch")
- public String fetch(Client client) throws Exception {
- var conn = client.getConnection("my_api");
- // implement task logic
- return result;
- }
-
- @Airflow.TaskHandler(dag = "my_dag", task = "process")
- public long process(Client client, String fetched) {
- var threshold = (String) client.getVariable("process_threshold");
- // implement task logic
- return count;
- }
- }
-
-A task method may declare ``throws Exception``; any uncaught exception causes
the task instance to be marked
-as failed in Airflow (triggering retries if configured on the stub).
+.. code-block:: java
-.. _java-sdk/interface-api:
+ var dag = new DagDef("java_etl").config("schedule",
"@daily").config("catchup", false);
+ dag.task("load", Load.class).config("retry_delay",
Duration.ofMinutes(5)).config("queue", "java");
-Interface-based API
-~~~~~~~~~~~~~~~~~~~
+Durations and date-times are ISO-8601 strings in annotations, and
``java.time.Duration`` and
+``java.time.OffsetDateTime`` values in ``config``. An unknown setting or a
value of the wrong type fails the
+build for an annotation, and the ``config`` call for an object.
-Implement the ``Task`` interface directly for full control over how tasks are
registered and how XComs are
-read. Each task is registered as a ``TaskDef`` on a ``DagDef``; both carry a
fluent
-``config(key, value)`` whose keys are Airflow's own setting names.
+A cron ``schedule`` runs in the time zone of ``startDate``, or in UTC without
one.
-The runner creates a fresh instance of the task class through reflection for
every task-instance run,
-which puts four constraints on the class:
+Every task needs a queue that :ref:`queue_to_coordinator
<java-sdk/coordinator-config>` sends to a
+``JavaCoordinator``. There is no Dag-level queue yet, so set ``queue`` on each
task.
Review Comment:
If this is true, many examples above are missing `queue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
