uranusjr commented on code in PR #74701:
URL: https://github.com/apache/airflow/pull/74701#discussion_r4335600697
##########
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.
-* The task class itself must be ``public``.
-* It must be concrete: not abstract and not an interface.
-* It must declare a public no-argument constructor.
-* If nested inside another class, it must be a ``static`` nested class.
+Order-only dependencies
+~~~~~~~~~~~~~~~~~~~~~~~
-A class that violates any of these fails at runtime with a ``Cannot
instantiate task class`` error in the
-task log.
+When a task only has to run after another, without receiving its value, draw
the dependency with ``before``
+and ``after``, the Java spelling of Python's ``>>`` and ``<<``. Both work the
same in the two styles:
.. code-block:: java
- import org.apache.airflow.sdk.*;
+ rows.before(audit()); // rows >> audit
+ load.after(transform, validate); // [transform, validate] >> load
- public class FetchTask implements Task {
- @Override
- public void execute(Context context, Client client) throws Exception {
- var conn = client.getConnection("my_api");
- // implement task logic
- client.setXCom(result);
- }
- }
+Both take any number of tasks and return the task they were called on, so
calls chain. ``Flow.of(a, b).before(c,
+d)`` draws every edge between two sets in one call. ``Flow`` is
``org.apache.airflow.sdk.Deps.Flow``, and needs
+no import inside the wiring class. An edge to a task of another Dag, or one
that closes a
+cycle, fails when the Dag is registered.
-Implement ``InputTask<I>`` instead when the Python Dag calls the stub with
TaskFlow arguments: the SDK
-resolves them from the call site and passes them in. The type argument is a
``TaskInput`` whose public
-fields declare what the task expects. See :ref:`java-sdk/arg-binding`.
+Task groups
+~~~~~~~~~~~
-Register each task against the Dag and task the Python file declares.
``register`` is one
-overloaded verb: these ids, or the class an annotated handler lives on.
+A task group gathers tasks that the Airflow UI shows as one node, as Python's
``TaskGroup`` does. Every id
+declared in a group is prefixed with the group's id.
+
+With annotations, a ``static`` nested ``@Airflow.TaskGroup`` class holds the
tasks of one group, and nesting
+one group class in another nests the groups:
.. code-block:: java
- public class MyBundle {
- public static class ProcessInput implements TaskInput {
- public String fetched;
- }
+ @Airflow.TaskGroup // group "Staging", from the
class name
+ static class Staging {
+ @Airflow.Task
+ public long stage(long rows) { ... } // task "Staging.stage"
- public static class ProcessTask implements InputTask<ProcessInput> {
- @Override
- public void execute(Context context, Client client, ProcessInput
input) throws Exception {
- // implement task logic
- client.setXCom(input.fetched);
- }
+ @Airflow.TaskGroup(id = "checks")
+ static class Checks {
+ @Airflow.Task
+ public void nulls(long staged) { ... } // task "Staging.checks.nulls"
}
+ }
- public static void main(String[] args) {
- var bundle = new Bundle()
- .register("my_dag", "fetch", FetchTask.class)
- .register("my_dag", "process", ProcessTask.class);
- Server.create(args).serve(bundle);
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ var staged = staging().stage(extract());
+ staging().checks().nulls(staged);
}
}
-A task class can be top-level like ``FetchTask``, or a nested ``static`` class
like ``ProcessTask``.
-Place them under the standard ``src/main/java/<package>/`` source tree, and set
-``airflowBundle.mainClass`` to the class that provides ``main``. From that
point onward, both APIs
-use the same ``./gradlew bundle`` command and deploy the resulting
``build/bundle/`` directory in the same way.
+With the interface, ``taskGroup`` declares a group on the Dag or on another
group:
-See the `Java SDK API Reference
<https://airflow.apache.org/docs/java-sdk/stable/>`__ for more details.
+.. code-block:: java
-.. _java-sdk/arg-binding:
+ var staging = dag.taskGroup("staging");
+ var stage = staging.task("stage", Stage.class); //
"staging.stage"
+ staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); //
"staging.checks.nulls"
+ extract.before(staging);
-Binding stub arguments
-~~~~~~~~~~~~~~~~~~~~~~
+A group takes ``before`` and ``after`` too, which orders the whole group. A
group id is made of ASCII letters,
+digits, underscores and dashes, and no task or other group in the Dag can
share it.
-Calling a ``@task.stub`` TaskFlow-style in the Python Dag is what declares the
graph, and the
-supervisor delivers the resulting argument bindings to the Java runtime with
every task run. A
-binding carries either an upstream task's ``return_value`` XCom or an inline
literal written at the
-call site.
+Conditional branching
+~~~~~~~~~~~~~~~~~~~~~
-Positional binding
-^^^^^^^^^^^^^^^^^^
+A condition is a task that returns a boolean, and names the task each outcome
runs. The side not taken is
+skipped.
-A task method's data parameters bind **by position**, in declaration order —
the injected ``Client``
-and ``Context`` parameters do not take up a position. Java parameter names
are not part of the API,
-so renaming one in an IDE never rebinds an input.
+With annotations, mark the method with ``@Airflow.If``, and name the sides in
the wiring class:
-.. code-block:: python
+.. code-block:: java
- @task.stub(queue="java")
- def score(rows, threshold): ...
+ @Airflow.If
+ public boolean hasRows(long transformed) {
+ return transformed > 0;
+ }
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ var transformed = transform(extract(), lit(0.9));
+ hasRows(transformed).Then(load(transformed)).Else(reportEmpty());
+ }
+ }
- score(load_rows(), 0.75)
+With the interface, implement ``ConditionTask`` and register it with
``dag.If``:
.. code-block:: java
- @Airflow.TaskHandler(dag = "etl", task = "score")
- public long score(Client client, long rows, double threshold) {
- // rows <- the load_rows XCom (position 0)
- // threshold <- the literal 0.75 (position 1)
+ public class HasRows implements ConditionTask {
+ @Override
+ public boolean decide(Context context, Client client) {
+ return ((Number) client.getXCom("transform")).longValue() > 0;
+ }
}
-A primitive parameter cannot hold ``null``, so the task fails with
``MissingXComException`` when its
-binding resolves to nothing; declare a boxed type (``Long``, ``Double``, …) to
receive ``null``
-instead. The method must declare as many data parameters as the call site
bound: positions carry
-the whole meaning of a flat binding, so any other count has already shifted
them, and the task fails
-rather than running on arguments it has mistaken for others. A parameter the
Python call omitted
-does not count towards that. Its default still arrives, but a method that
does not declare it is
-not reading shifted arguments, so the SDK drops it before comparing the two
counts.
-
-Generic parameters are decoded element by element. Declare ``List<Double>``
and ``values.get(0)``
-really is a ``Double``, even though the call site passed whole numbers and the
wire carries them as
-integers. Without that the list would hold ``Long`` values while claiming to
hold ``Double``, and
-the ``ClassCastException`` would land on whichever line first read an element
rather than on the
-binding that got it wrong.
-
-Named binding with a ``TaskInput``
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-To bind keyword arguments by name, declare a class implementing ``TaskInput``.
Each of its public
-non-final fields receives the argument whose name matches it, **ignoring case
and underscores**, so
-the stub's ``snake_case`` arguments reach ``camelCase`` Java fields with
nothing declared. That is
-the same fold the Go and TypeScript SDKs apply, so one Python signature binds
identically in every
-SDK.
-
-The class needs a public no-argument constructor. A name mismatch in either
direction is logged
-rather than failed, because a field binds by name: a field nothing supplies
keeps its Java default,
-and an argument no field claims changes nothing the task reads. Once a field
has claimed its
-argument, an argument that resolves to nothing is a value and not a mistake,
so a boxed or reference
-field takes ``null`` and a primitive field fails.
-
-No two fields may claim argument names that differ only in case or
underscores; that fails when the
-bundle is built, because the fold cannot tell them apart. Two *arguments*
that collide that way
-reach only a field naming one of them exactly, and a field that would match
both is left unfilled
-and reported rather than handed the wrong value.
+ dag.If(HasRows.class).after(transform).Then(load).Else(reportEmpty);
-.. code-block:: python
+``Else`` is optional.
- @task.stub(queue="java")
- def score(region_code, threshold): ...
+Only the task on the side not taken is skipped, and it stays skipped if it is
cleared later. A task that runs
+after both sides needs a trigger rule that accepts a skipped upstream, such as
``none_failed_min_one_success``.
+Multi-way branching
+~~~~~~~~~~~~~~~~~~~
- score(region_code="emea", threshold=load_threshold())
+A switch is a task that chooses one of several tasks to run, and skips every
other one. It returns the class
+of the task to run.
-.. code-block:: java
+With annotations, mark the method with ``@Airflow.Switch``, return a class the
annotation processor generated
+for a task of the Dag, and list the cases in the wiring class:
- public static class ScoreInput implements TaskInput {
- // Pinned so the field can be called region. Or drop the annotation and
- // write: public String regionCode;
- @ArgName("region_code")
- public String region;
+.. code-block:: java
- public double threshold; // binds threshold
+ @Airflow.Switch
+ public Class<? extends Task> pickReport(long transformed) {
+ return transformed > 100 ? EtlPipelineBuilder.ReportLong.class :
EtlPipelineBuilder.ReportShort.class;
}
- @Airflow.TaskHandler(dag = "etl", task = "score")
- public long score(Client client, ScoreInput input) { ... }
-
-Reach for ``@ArgName`` when the argument name is not a legal or usable Java
identifier — a Python
-keyword such as ``class``, say — or when the field should read differently
from the argument. A
-pinned name is matched as written, with no folding.
-
-A task method declares flat data parameters **or** one ``TaskInput``, never
both, so field names and
-flat positions cannot shift each other. Mixing them, or declaring two, fails
the build.
-
-Binding in the interface-based API
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ pickReport(transform(extract(),
lit(0.9))).Case(reportLong()).Case(reportShort());
+ }
+ }
-A task written against the interface has no parameter list to bind, so it
declares its input as the
-type argument of ``InputTask<I>`` instead — the same ``TaskInput`` an
annotated task would declare,
-binding the same way:
+With the interface, implement ``SwitchTask`` and register it with
``dag.Switch``:
.. code-block:: java
- public class ScoreTask implements InputTask<ScoreInput> {
+ public class PickReport implements SwitchTask {
@Override
- public void execute(Context context, Client client, ScoreInput input)
throws Exception {
- // input.region, input.threshold
+ public Class<? extends Task> choose(Context context, Client client) {
+ return ((Number) client.getXCom("transform")).longValue() > 100 ?
ReportLong.class : ReportShort.class;
}
}
-A ``TaskInput`` is the only way an interface task receives bound values; there
is no positional form.
-A positional read is safe in the code the annotation processor writes, because
it type-checks each
-position against the method signature it serves, and is the wrong thing to ask
of code a person
-writes and later reads. A call site with a single argument is worth the
one-field class.
+
dag.Switch(PickReport.class).after(transform).Case(reportLong).Case(reportShort);
-An ``InputTask`` whose type argument is not a concrete ``TaskInput`` fails
when the bundle is built,
-rather than mid-run. Plain ``Task`` remains the right interface for a task
the Dag file
-calls with no arguments.
+Exactly one case runs. A switch that returns a class that is not one of its
cases fails, and two cases cannot
+run the same class. A task that runs after several cases needs a trigger rule
that accepts a skipped upstream,
+such as ``none_failed_min_one_success``.
-.. _java-sdk/native-dags:
+Edge labels
+~~~~~~~~~~~
-Native Java Dags
-----------------
+``Label`` puts a label on an edge, which the Airflow UI shows on the graph, as
Python's ``Label`` does.
+``Label.of(text).on(...)`` goes in ``before`` or ``after`` in place of the
task or group at the other end of the edge,
+and works the same in the two styles:
-A Dag can also be authored entirely in Java: the annotations (or the
``DagDef`` / ``TaskDef``
-objects) carry the configuration, and Java declares the graph.
+.. code-block:: java
-Building the Dag in Java
-~~~~~~~~~~~~~~~~~~~~~~~~
+ loaded.before(Label.of("when empty").on(reportEmpty)); //
load >> Label("when empty") >> report_empty
+ cleaned.after(Label.of("always").on(loaded)); //
load >> Label("always") >> cleanup
+ loaded.before(Label.of("forks").on(reportEmpty, reportSuspicious)); //
one label on two edges
-``dag.task(...)`` registers a task as it creates it and hands back a handle.
``before`` and
-``after`` draw every edge on this surface, and the task body moves the data
itself, by reading the
-upstream's XCom through ``Client``:
+A label goes only on the edges between the receiver of ``before`` or ``after``
and the tasks passed to ``on``, so each
+edge of a fan-out can carry its own label. Declaring an edge that already
exists only adds the label, which is how the
+sides of a condition get theirs. A new label replaces the old one:
.. code-block:: java
- var dag = new DagDef("java_etl");
+ var hasRows =
dag.If(HasRows.class).after(transform).Then(load).Else(reportEmpty);
+ hasRows.before(Label.of("rows found").on(load), Label.of("no
rows").on(reportEmpty));
- var extract = dag.task("extract", Extract.class);
- var transform = dag.task("transform", Transform.class);
- var load = dag.task("load", Load.class);
+``Label`` is ``org.apache.airflow.sdk.Label``. Unlike ``Flow``, the wiring
class does not inherit it, so it needs an
+import. Pass the value ``on`` returns only to ``before`` and ``after``:
calling ``before`` or ``after`` on it fails. On
+an edge to or from a task group, the UI shows the label on the group's node.
- transform.after(extract).before(load);
+A label never changes which tasks an edge connects. Python's ``Label`` can:
when the end on its left is in a task group
+and the other end is not in that group, Python replaces that end with the
group.
Review Comment:
We probably don’t need to mention this. (If anything, the Python docs should
mention this since the behavior is arguably not intuitive.)
--
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]
