FrankYang0529 commented on code in PR #74701: URL: https://github.com/apache/airflow/pull/74701#discussion_r4326644323
########## java-sdk/example/README.md: ########## @@ -0,0 +1,107 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + --> + +# Java SDK examples + +Example Dags for the Apache Airflow Java SDK. See the [Java SDK guide][guide] for the full documentation. + +[guide]: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/language-sdks/java.html + +## Build + +Publish the SDK to the local Maven repository, then build the example bundle into `build/bundle/`: + +```bash +cd java-sdk +./gradlew publishToMavenLocal -PskipSigning=true + +cd example +../gradlew bundle +``` + +## Configure Airflow + +Register `build/bundle/` as a Dag bundle, and route the `java` queue to the Java coordinator: + +```bash +export AIRFLOW__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": "/absolute/path/to/airflow/java-sdk/example/build/bundle"} + } +]' +export AIRFLOW__SDK__COORDINATORS='{ + "java": { + "classpath": "airflow.sdk.coordinators.java.JavaCoordinator", + "kwargs": {"task_handler_bundle_name": "java-task-handlers"} + } +}' +export AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"java": "java"}' +``` + +The workers and the Dag processor need Java 11 or later on `PATH`. + +## Dag definition + +These Dags are declared entirely in Java, in `src/java/org/apache/airflow/example/nativedag/`: + +| Dag | Source | Shows | +| -------------------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------ | +| `java_native_annotation_example` | `AnnotationExample.java` | The annotation style: task inputs, a task group, `@Airflow.If`, `@Airflow.Switch`, and `TriggerDagRun` | +| `java_native_interface_example` | `InterfaceExample.java` | A similar Dag in the interface style, with `DagDef` | +| `java_native_target_example` | `TargetExample.java` | The Dag the other two trigger | Review Comment: No task in `nativedag/*.java` sets a `queue`. We need to add it, so users can follow the example to run the dag. ########## 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. -Both are variadic, so ``a.before(b, c)`` fans out and ``d.after(b, c)`` fans in, and both return -their own receiver, so a chain reads from one task outwards. ``Flow.of(a, b).before(c, d)``, from -``org.apache.airflow.sdk.Deps.Flow``, draws every edge between two sets in one call. - -Edges are checked when the Dag is registered with a ``Bundle``: an upstream that belongs to another -Dag, or to no Dag, and a cycle anywhere in the graph both fail there rather than at the first task -run. - -Wiring the graph with ``@Airflow.Deps`` -~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +Triggering another Dag +~~~~~~~~~~~~~~~~~~~~~~ -For a Dag written with annotations, the graph is declared by a nested ``@Airflow.Deps`` class. The -annotation processor generates a ``<ClassName>Deps`` interface, the *wiring view*, with one method -per ``@Airflow.Task`` method: the injected ``Client`` and ``Context`` parameters are dropped, each -data parameter becomes an ``Arg<T>``, and the return value becomes a ``TaskRef<T>``. Calling a view -method registers its task, and passing the handle one returned into another feeds the upstream's -output into the downstream's parameter *and* wires the data edge. The call graph is the task graph, -and ``javac`` type-checks it. +``TriggerDagRun`` makes a task that starts a run of another Dag, as Python's ``TriggerDagRunOperator`` does. -Declare the wiring class as a ``static`` nested class of the Dag class that ``implements`` the -generated view, with a no-argument ``depends()`` method: +With annotations, a ``@Airflow.Task`` method returns one. The method runs when the Dag is built, not when the +task runs, so it takes no arguments: .. code-block:: java - @Airflow.Dag( - id = "java_etl", - schedule = "@daily", - description = "Pure-Java Dag built with annotations", - tags = {"example", "java-sdk"}) - public class EtlPipeline { - - @Airflow.Task(id = "extract", retries = 2) - public long extract() { - return 42L; - } - - @Airflow.Task(id = "transform") - public long transform(long extracted, double factor) { - return (long) (extracted * factor); - } + @Airflow.Task + public TriggerDagRun triggerDownstream() { Review Comment: Do we need to add `queue` here like other tasks? -- 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]
