Rédaction de tâches avec le Java SDK

Le Java SDK Airflow implémente le modèle language-SDK pour la JVM : votre DAG reste en Python, et chaque instance de tâche s'exécute dans un sous-processus JVM éphémère. Cette skill couvre l'API native spécifique à Java. Le modèle partagé — le pattern Python @task.stub, l'appariement d'IDs et le contrat XCom-as-JSON — se trouve dans authoring-language-sdk-tasks ; lisez-le d'abord si vous découvrez les language SDKs.

Expérimental. Le Java SDK est en aperçu. Les coordonnées des artefacts et les APIs peuvent changer.

Skills connexes : authoring-language-sdk-tasks (stub Python partagé + concepts), configuring-airflow-language-sdks (acheminer la queue vers JavaCoordinator), deploying-java-sdk-bundles (compiler et livrer le JAR).

Récapitulatif : le côté Python

Les tâches Java sont associées à des stubs Python qui ne contiennent aucune logique — ils déclarent la tâche, la queue, le graphe de dépendances et les retentatives. Les IDs doivent correspondre exactement aux annotations Java, et un argument upstream sur un stub déclare seulement la dépendance (la valeur est récupérée en Java). Les règles complètes sont dans authoring-language-sdk-tasks ; la forme minimale :

from airflow.sdk import dag, task


@dag
def sales_pipeline():                     # dag_id "sales_pipeline" -> @Builder.Dag(id="sales_pipeline")
    @task.stub(queue="java")
    def extract(): ...                    # task_id "extract" -> @Builder.Task(id="extract")

    @task.stub(queue="java")
    def transform(extracted): ...

    transform(extract())


sales_pipeline()

Côté Java : deux APIs

Les deux APIs produisent un comportement identique à l'exécution ; choisissez selon votre style, et vous pouvez les mélanger dans un bundle.

API basée sur les annotations (recommandée)

Annotez une classe simple ; un processeur d'annotations génère le câblage (<ClassName>Builder) au moment de la compilation.

import static java.lang.System.Logger.Level.INFO;
import org.apache.airflow.sdk.*;

@Builder.Dag(id = "sales_pipeline")          // doit correspondre au dag_id Python
public class SalesPipeline {
  private static final System.Logger log = System.getLogger(SalesPipeline.class.getName());

  @Builder.Task(id = "extract")              // doit correspondre au nom @task.stub Python
  public long extract(Client client) {
    var conn = client.getConnection("sales_db");
    log.log(INFO, "connected to {0}", conn.host);
    return 42L;                              // la valeur de retour est poussée comme return_value XCom
  }

  @Builder.Task(id = "transform")
  public long transform(
      Client client,
      @Builder.XCom(task = "extract") long recordCount) {  // récupère return_value d'extract
    var threshold = (String) client.getVariable("transform_threshold");
    return recordCount * 2;
  }

  @Builder.Task   // id omis -> le nom de méthode "load" est utilisé
  public void load(Context context, @Builder.XCom(task = "transform") long transformed) {
    log.log(INFO, "attempt {0}, value {1}", context.ti.tryNumber, transformed);
  }
}

Référence des annotations :

La valeur de retour d'une méthode de tâche est automatiquement poussée comme le XCom return_value de cette tâche. Une méthode peut déclarer throws Exception ; toute exception non capturée échoue l'instance de tâche (ce qui déclenche les retentatives si le stub les a configurées).

API basée sur les interfaces

Implémentez Task directement quand vous voulez un contrôle total sur l'enregistrement et la gestion des XComs.

import org.apache.airflow.sdk.*;

public class ExtractTask implements Task {
  @Override
  public void execute(Context context, Client client) throws Exception {
    var conn = client.getConnection("sales_db");
    // ... faire le travail ...
    client.setXCom(42L);   // pousser return_value explicitement
  }
}

Enregistrez les tâches manuellement dans un Dag et exposez-le via un BundleBuilder :

public class MyBundle implements BundleBuilder {
  @Override
  public Iterable<Dag> getDags() {
    var dag = new Dag("sales_pipeline");      // l'ID DAG correspond à Python
    dag.addTask("extract", ExtractTask.class);
    dag.addTask("transform", TransformTask.class);
    return java.util.List.of(dag);
  }
}

Chaque classe Task a besoin d'un constructeur public sans argument. Les IDs de tâche doivent être uniques dans un DAG, et les IDs de DAG uniques dans un bundle.

Le point d'entrée

Chaque bundle a un main qui confie vos DAGs au serveur SDK. Le serveur se connecte au coordinator, exécute une instance de tâche et se termine.

import java.util.List;
import org.apache.airflow.sdk.*;

public class Main implements BundleBuilder {
  @Override
  public Iterable<Dag> getDags() {
    // Avec l'API annotation, les classes *Builder sont générées à la compilation.
    return List.of(SalesPipelineBuilder.build());
  }

  public static void main(String[] args) {
    Server.create(args).serve(new Main().build());
  }
}

Server.create(args) analyse les détails de connexion qu'Airflow transmet en ligne de commande — ne les construisez pas à la main. Enregistrez cette classe main comme classe principale du bundle quand vous le compilez (voir deploying-java-sdk-bundles).

Communiquer avec Airflow depuis une tâche : Client

Un Client est passé dans chaque tâche et a une portée limitée à la DAG run et l'instance de tâche actuelles.

Context

Le paramètre Context expose les métadonnées de run : context.dagRun (dagId, runId) et context.ti (dagId, runId, taskId, mapIndex, tryNumber). tryNumber est utile pour la logique consciente des retentatives.

XCom : types Java

Les XComs franchissent la limite en JSON (le contrat partagé est dans authoring-language-sdk-tasks). Quand vous en relisez un en Java vous obtenez :

Déclarez les types de paramètres @Builder.XCom pour correspondre. Une non-correspondance (par exemple déclarer int quand la valeur est une String) échoue la tâche.

Enregistrement des logs

Déclarez un logger comme champ statique nommé d'après la classe — le pattern conventionnel indépendamment du framework :

private static final System.Logger log = System.getLogger(SalesPipeline.class.getName());

Pour que les enregistrements atteignent le magasin de logs de tâche Airflow (et s'affichent dans l'UI), le bundle doit inclure un des artefacts d'intégration de logging du SDK (airflow-sdk-jpl, airflow-sdk-slf4j, airflow-sdk-log4j2 ou airflow-sdk-jul). Les dépendances et la configuration par framework se trouvent dans la section d'intégration de logging de deploying-java-sdk-bundles. System.Logger (JPL) avec airflow-sdk-jpl est l'option la plus légère et ne nécessite aucune configuration.

Un exemple complet et fonctionnel est livré avec le SDK

Le repository SDK inclut un exemple exécutable sous java-sdk/example/ :

• src/resources/dags/java_examples.py — DAGs Python associant des tâches Python avec des stubs Java, incluant un stub load avec retries=1.
• src/java/.../AnnotationExample.java — API annotation, incluant une tâche qui échoue sur tryNumber == 1 et réussit à la retentative.
• src/java/.../InterfaceExampleBuilder.java — les mêmes tâches via l'interface Task et Dag.addTask(...).
• src/java/.../ExampleBundleBuilder.java — un BundleBuilder retournant les deux DAGs plus le point d'entrée main.

Pointez les utilisateurs vers celui-ci pour une référence bout à bout.

Pièges spécifiques à Java

• Castez intentionnellement les retours Object. getVariable et getXCom retournent Object ; faites correspondre le cast au type JSON (voir le tableau ci-dessus).
• Les types de paramètres @Builder.XCom doivent correspondre au type JSON stocké, sinon la tâche échoue à l'exécution.
• Le processeur d'annotations doit être dans la compilation pour l'API annotation (génère <ClassName>Builder) ; il n'est pas nécessaire pour l'API interface. Voir deploying-java-sdk-bundles.
• Consultez authoring-language-sdk-tasks pour les pièges agnostiques du langage (appariement d'IDs, une JVM par instance de tâche, queue/retentatives sur le stub).

Skills connexes

• authoring-language-sdk-tasks : Pattern et concepts Python-stub partagés (à lire d'abord).
• configuring-airflow-language-sdks : Acheminer la queue java vers JavaCoordinator et définir les options JRE/coordinator.
• deploying-java-sdk-bundles : Compiler le bundle (Gradle/Maven) et placer le JAR où Airflow peut le trouver.
• authoring-dags : Rédaction générale de DAGs Airflow.