configuring-airflow-language-sdks

为语言 SDK 配置 Airflow

要运行语言 SDK 任务，Airflow 需要知道两件事：由哪个协调器启动原生子进程，以及由哪个队列将任务路由到该协调器。所有语言 SDK 使用的机制都相同——只有各协调器的 classpath 和 kwargs 不同。本指南会先介绍通用配置，再介绍各协调器的选项。本指南不依赖特定平台：开源 Airflow 和 Astro 等托管平台适用相同的设置。

实验性功能。 语言 SDK 处于预览阶段；配置键可能会更改。

有关任务代码，请参阅 authoring-language-sdk-tasks（以及各语言的编写指南，例如 authoring-java-sdk-tasks、authoring-go-sdk-tasks）。有关构建和分发制品，请参阅各语言的部署指南（例如 deploying-java-sdk-bundles、deploying-go-sdk-bundles）。

---

Worker 的先决条件

• Worker 节点上必须有 SDK 所需的运行时或制品，因为协调器会为每个任务实例启动一个原生子进程。具体要求因 SDK 而异——请参阅各协调器的选项（Java SDK 需要 JRE 17+；Go SDK 无需语言运行时——该 bundle 是自包含的原生可执行文件，但必须针对 Worker 的操作系统和架构构建）。
• 编译后的/原生制品必须能在 Worker 上访问。请参阅相应语言的部署指南。
• 协调器随 Airflow Task SDK 一起提供（随 Airflow 安装的 apache-airflow-task-sdk）。无需安装额外的 Python 包。

---

两项设置

两项设置都位于 [sdk] 配置段中，适用于所有语言 SDK：

1. coordinators——一个 JSON 对象，将你选择的协调器名称映射到其实现（classpath）和构造函数 kwargs。
2. queue_to_coordinator——一个 JSON 对象，将任务队列映射到协调器名称。

如果任务的存根设置了 queue="..."，任务就会交给指定的协调器，由其启动原生子进程。协调器名称可以任意指定——只需确保两项设置中的字符串相同。队列名称必须与 Python @task.stub 上设置的 queue= 一致。

选项 A：airflow.cfg

[sdk]
coordinators = {
  "java-jdk17": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {"jars_root": ["/opt/airflow/jars"]}
  },
  "go": {
    "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator",
    "kwargs": {"executables_root": ["/opt/airflow/executable-bundles"]}
  }
}
queue_to_coordinator = {"java": "java-jdk17", "golang": "go"}

选项 B：环境变量

每个值都必须是有效的单行 JSON。这种形式适用于容器、.env 文件、Docker Compose 和 Helm。

export AIRFLOW__SDK__COORDINATORS='{"java-jdk17": {"classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": {"jars_root": ["/opt/airflow/jars"]}}, "go": {"classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs": {"executables_root": ["/opt/airflow/executable-bundles"]}}}'
export AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{"java": "java-jdk17", "golang": "go"}'

以上示例会同时注册多个协调器（每种语言一个），并将不同队列映射到各个协调器——只需注册你会用到的协调器。

---

各协调器的选项

classpath 和 kwargs 因协调器而异。随着新的语言 SDK 发布，可在此处添加相应小节。

JavaCoordinator

• classpath：airflow.sdk.coordinators.java.JavaCoordinator
• Worker 运行时：JRE 17+（PATH 中有 java，或设置 java_executable）。

参数	默认值	描述
jars_root	(必填)	递归扫描 .jar 文件的一个或多个目录。接受字符串或字符串/路径列表。类路径会自动组装。
java_executable	"java"	java 二进制文件的路径。默认为 $PATH 中的 java。
jvm_args	[]	额外的 JVM 参数，例如 ["-Xmx1g", "-Dsome.property=value"]。
main_class	(自动检测)	显式指定入口类。如果省略，协调器会扫描 jars_root，查找清单中声明了 Main-Class 的 JAR。如果存在多个可执行 JAR，请显式设置此项，否则选择结果不确定。
task_startup_timeout	10.0	启动后等待子进程连接的秒数。如果 JVM 启动较慢（硬件受限、类路径较大、首次冷启动），请增大此值。

通过 java.util.logging 进行 Java 日志记录。 在 SDK 的日志集成中，只有 JPL 和 SLF4J 是零配置构建依赖项；Log4j 2 和 JUL 需要额外设置，详见 deploying-java-sdk-bundles 中的日志集成部分。JUL 文档中提到的替代方案（无需在 main() 中调用 AirflowJulHandler.setup()）是使用 logging.properties 文件，并通过 jvm_args 配置：

[sdk]
coordinators = {
  "java-jdk17": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {
      "jars_root": ["/opt/airflow/jars"],
      "jvm_args": ["-Djava.util.logging.config.file=/opt/airflow/logging.properties"]
    }
  }
}

ExecutableCoordinator（Go 和其他自包含可执行文件 SDK）

• classpath：airflow.sdk.coordinators.executable.ExecutableCoordinator
• Worker 运行时：除 bundle 本身外无需其他运行时。该 bundle 是自包含的原生可执行文件（AFBNDL01），因此无需语言运行时，但必须针对 worker 的操作系统/架构构建（不匹配会导致 exec format error）。

参数	默认值	描述
executables_root	(必填)	递归扫描可执行 bundle（带有 AFBNDL01 尾标记的原生二进制文件）的一个或多个目录。接受字符串或字符串/路径列表。bundle 根据尾标记魔数识别，而非文件名。协调器会将传入的 dag_id 与各个 bundle 中嵌入的清单进行匹配，并在启动前验证其完整性哈希。
task_startup_timeout	10.0	启动后等待子进程连接的秒数。如果 bundle 启动较慢（硬件受限、首次冷启动），请增大此值。

（未来其他语言的协调器会在此列出各自的 classpath、运行时和 kwargs。）

---

验证配置

1. 确认运行时/制品在 worker 运行的环境中可用：对于 Java SDK，通过 astro dev bash 或 docker compose exec ... 运行 java -version；对于 Go SDK，确认打包后的 bundle 存在且与 worker 的操作系统/架构匹配。
2. 确认 kwargs 中引用的制品目录（例如 jars_root、executables_root）在 worker 文件系统中确实包含你的制品。
3. 触发 DAG 并打开原生任务的日志——你应能看到子进程启动及任务输出。

故障排查

症状	可能原因 / 解决方法
任务立即失败，提示 coordinator 或 queue	coordinators / queue_to_coordinator 不是有效的单行 JSON，或者队列名称与存根的 queue= 不匹配。修正 JSON 并重启。
找不到运行时（例如 java: command not found）	工作节点上没有该语言的运行时，或者可执行文件路径 kwarg 有误。安装运行时并验证其版本。
“No artifact found” / “no DAGs” / “no bundle contains dag_id”	artifact-directory kwarg 指向了错误位置、artifact 尚不存在，或者其 dag_id 与存根不匹配。确认路径和 ID。
入口点错误或不明确（Java）	jars_root 下有多个可执行 JAR。显式设置 main_class。
Go bundle 被静默跳过	不是有效的 AFBNDL01 bundle，或者其完整性哈希验证失败（任何剥离 / 签名 / 重建操作后都要重新打包）。
Go bundle 出现 exec format error	构建时使用的 OS / 架构与工作节点不同。使用 --goos / --goarch 交叉编译（参见 deploying-go-sdk-bundles）。
DAG 运行卡在原生任务	增大 task_startup_timeout（例如 30.0）；首次运行时子进程启动可能较慢。

---

相关技能

• authoring-language-sdk-tasks：共享的 Python 存根模式和概念模型。
• authoring-java-sdk-tasks：Java 任务代码及其对应的 Python 存根。
• deploying-java-sdk-bundles：构建 bundle，并将 artifact 放到 coordinator 扫描的位置。
• authoring-go-sdk-tasks：Go 任务代码及其对应的 Python 存根。
• deploying-go-sdk-bundles：构建 / 打包 Go bundle，并将其放到 coordinator 扫描的位置。
• deploying-airflow：在 Astro、Docker Compose 或 Kubernetes 上部署 Airflow 的通用方法。