authoring-java-sdk-tasks

编写 Java SDK 任务

Airflow Java SDK 为 JVM 实现了语言 SDK 模型：DAG 保留在 Python 中，每个任务实例都在一个短生命周期的 JVM 子进程中运行。本技能介绍 Java 专属 原生 API。共用模型——Python @task.stub 模式、ID 匹配和 XCom 作为 JSON 的约定——详见 authoring-language-sdk-tasks；如果你刚接触语言 SDK，请先阅读该文档。

实验性功能。 Java SDK 目前处于预览阶段。构件坐标和 API 可能会变更。

相关技能： authoring-language-sdk-tasks（共用的 Python 存根与概念）、configuring-airflow-language-sdks（将队列路由到 JavaCoordinator）、deploying-java-sdk-bundles（编译并打包 JAR）。

---

回顾：Python 侧

Java 任务与不含逻辑的 Python 存根配对使用——它们声明任务、队列、依赖图和重试。ID 必须与 Java 注解完全一致，存根中的上游参数仅用于声明依赖（值由 Java 获取）。完整规则见 authoring-language-sdk-tasks；最简结构如下：

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()

---

Java 侧：两种 API

两种 API 的运行时行为完全相同；可按风格选择，也可以在同一个 bundle 中混用。

基于注解的 API（推荐）

为普通类添加注解；注解处理器会在编译时生成接线代码（<ClassName>Builder）。

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

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

  @Builder.Task(id = "extract")              // must match the Python @task.stub name
  public long extract(Client client) {
    var conn = client.getConnection("sales_db");
    log.log(INFO, "connected to {0}", conn.host);
    return 42L;                              // return value is pushed as the return_value XCom
  }

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

  @Builder.Task   // id omitted -> the method name "load" is used
  public void load(Context context, @Builder.XCom(task = "transform") long transformed) {
    log.log(INFO, "attempt {0}, value {1}", context.ti.tryNumber, transformed);
  }
}

注解参考：

注解	用途
@Builder.Dag(id = "...")	将类标记为任务容器。id 必须与 Python dag_id 一致；若省略，则使用类名。可选的 to = "..." 用于重命名生成的 builder（默认为 <ClassName>Builder）。
@Builder.Task(id = "...")	将方法标记为任务。id 必须与 Python @task.stub 函数名一致；若省略，则使用方法名。
@Builder.XCom(task = "...", key = "...")	将上游任务的 XCom 注入为参数。task 默认为参数名；key 默认为生成该 XCom 的任务的 return_value。参数类型必须与所存储的 JSON 值兼容。

任务方法的返回值会自动作为该任务的 return_value XCom 推送。方法可以声明 throws Exception；任何未捕获的异常都会导致任务实例失败（如果存根配置了重试，则会触发重试）。

基于接口的 API

需要完全控制注册和 XCom 处理时，可直接实现 Task。

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");
    // ... do work ...
    client.setXCom(42L);   // push return_value explicitly
  }
}

在 Dag 中手动注册任务，并通过 BundleBuilder 暴露该 DAG：

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

每个 Task 类都需要一个公共的无参构造函数。一个 DAG 内的任务 ID 必须唯一，一个 bundle 内的 DAG ID 也必须唯一。

---

入口点

每个 bundle 都有一个 main，用于将 DAG 交给 SDK 服务。该服务会连接协调器，运行一个任务实例，然后退出。

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

public class Main implements BundleBuilder {
  @Override
  public Iterable<Dag> getDags() {
    // With the annotation API, the *Builder classes are generated at compile time.
    return List.of(SalesPipelineBuilder.build());
  }

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

Server.create(args) 会解析 Airflow 通过命令行传入的连接详情——不要手动构造这些详情。构建 bundle 时，请将此 main 类记录为 bundle 的主类（参见 deploying-java-sdk-bundles）。

在任务中与 Airflow 通信：Client

每个任务都会传入一个 Client，其作用范围限定为当前 DAG 运行和任务实例。

调用	返回值	说明
client.getConnection(id)	Connection	字段：id、type、host、schema、login、password、port、extra。未设置的字段均为 null。如果连接不存在，则会抛出异常。
client.getVariable(key)	Object（或 null）	转换为预期类型，例如 (String) client.getVariable("threshold")。
client.getXCom(taskId)	Object（或 null）	默认读取另一个任务的 return_value。重载方法接受 key、dagId、runId、mapIndex 和 includePriorDates，用于跨 DAG／运行读取及读取映射任务。
client.setXCom(value)	—	推送 return_value XCom（接口 API）。值必须可序列化为 JSON。使用注解 API 时，只需返回一个值即可完成此操作。

Context

Context 参数会公开运行元数据：context.dagRun（dagId、runId）和 context.ti（dagId、runId、taskId、mapIndex、tryNumber）。tryNumber 可用于处理重试逻辑。

---

XCom：Java 类型

XCom 以 JSON 形式跨越边界（共享契约见 authoring-language-sdk-tasks）。在 Java 中读取时，得到的类型如下：

Python 类型	JSON	getXCom 返回的 Java 类型
int	integer	Long（过大时为 BigInteger）
float	decimal	Double
str	string	String
bool	boolean	Boolean
None	null	null
list	array	List<Object>
dict	object	Map<String, Object>

声明 @Builder.XCom 参数类型时，必须与之匹配。类型不匹配（例如，值为 String，却声明为 int）会导致任务失败。

---

日志记录

将日志记录器声明为一个以类名命名的静态字段——这是不受框架影响的惯用写法：

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

如需让日志记录显示在 Airflow 的任务日志存储中（并在 UI 中展示），Bundle 必须包含以下某个 SDK 日志集成工件：airflow-sdk-jpl、airflow-sdk-slf4j、airflow-sdk-log4j2 或 airflow-sdk-jul。依赖项和各框架的设置见 deploying-java-sdk-bundles 中的日志集成部分。使用 airflow-sdk-jpl 的 System.Logger（JPL）是最轻量的选项，无需配置。

---

SDK 附带完整的示例

SDK 仓库在 java-sdk/example/ 下提供了一个可运行示例：

• src/resources/dags/java_examples.py — Python DAG，其中将 Python 任务与 Java stub 配对，包括一个设置了 retries=1 的 load stub。
• src/java/.../AnnotationExample.java — 注解 API 示例，其中有一个任务会在 tryNumber == 1 时失败，并在重试时成功。
• src/java/.../InterfaceExampleBuilder.java — 通过 Task 接口和 Dag.addTask(...) 实现相同任务。
• src/java/.../ExampleBundleBuilder.java — 一个返回两个 DAG 和 main 入口点的 BundleBuilder。

可参考该示例了解端到端用法。

---

Java 特有的注意事项

• 有意转换 Object 返回值。 getVariable 和 getXCom 返回 Object；转换时请匹配 JSON 类型（见上表）。
• @Builder.XCom 参数类型必须与存储的 JSON 类型匹配，否则任务会在运行时失败。
• 使用注解 API 时，构建中必须包含注解处理器（它会生成 <ClassName>Builder）；接口 API 无需此处理器。参见 deploying-java-sdk-bundles。
• 与语言无关的注意事项（ID 匹配、每个任务实例启动一个 JVM、stub 上的队列／重试）请参见 authoring-language-sdk-tasks。

---

相关技能

• authoring-language-sdk-tasks：共享的 Python 存根模式和概念（请先阅读）。
• configuring-airflow-language-sdks：将 java 队列路由到 JavaCoordinator，并设置 JRE/协调器选项。
• deploying-java-sdk-bundles：构建 bundle（Gradle/Maven），并将 JAR 放在 Airflow 能找到的位置。
• authoring-dags：Airflow DAG 编写通用指南。