authoring-language-sdk-tasks

编写语言 SDK 任务（共享基础）

Airflow 语言 SDK 让你可以使用 Python 以外的语言实现任务逻辑，同时 DAG 和调度仍由 Python 管理。本技能介绍所有语言 SDK 通用的部分。每种语言都有对应的技能，介绍其原生 API、构建工具和运行时，详见各语言技能。

**实验性功能。**语言 SDK 目前处于预览阶段。API 和构件坐标可能会发生变化。

---

模型

DAG 照常使用 Python 编写。需要使用其他语言运行的任务会声明为存根，并路由到专用队列。运行时，Airflow 会将存根任务交给一个协调器，由它为该任务实例启动一个短生命周期的原生子进程，运行你编译后的原生代码，然后关闭该子进程。

适用于所有语言 SDK 的说明：

• 每个任务实例对应一个子进程——任务实例之间不共享进程内状态。请通过 XCom 或外部存储传递数据。
• **DAG、调度、重试和队列路由都由 Python 管理。**原生端只实现任务逻辑。
• **跨边界传递的数据格式为 JSON。**详见 XCom 即 JSON 契约。

---

双端模型

每个任务都有两部分，且两部分必须保持一致：

1. 普通 DAG 文件中的 Python 存根——不包含逻辑；它声明任务、队列、依赖关系图和重试策略。
2. 原生实现（Java、Go 等），其 ID 必须与 Python 端匹配，实际工作也在此处执行。

Python 端（调度）

下面的示例使用 Go SDK 进行说明，但 Python 端对所有语言 SDK 都完全相同。队列名称（此处为 "golang"）是你自行选择的任意标签，只需与 queue_to_coordinator 中的某个键匹配（参见 configuring-airflow-language-sdks）。请选择适合所路由到的 SDK 的名称。

from datetime import timedelta
from airflow.sdk import dag, task


@dag
def sales_pipeline():
    @task.stub(queue="golang")          # queue selects the coordinator (see configuring-airflow-language-sdks)
    def extract(): ...

    @task.stub(queue="golang")
    def transform(extracted): ...        # arg only declares the dependency

    @task.stub(queue="golang", retries=1, retry_delay=timedelta(seconds=5))
    def load(transformed): ...

    @task()                              # an ordinary Python task can sit downstream
    def report(loaded):
        print(f"done: {loaded}")

    report(load(transform(extract())))


sales_pipeline()

无论使用哪种语言，以下规则均适用：

• 存根函数名称就是任务 ID，而 @dag 名称（或 dag_id=）就是 DAG ID。原生端必须使用这些完全一致的 ID。
• 存根上的上游参数（例如 transform(extracted)）在 Python 中仅用于声明依赖关系。该值由原生端通过 XCom 获取——在 Python 中传入该值并不会将其交给原生代码。
• 队列、重试和其他任务参数都在存根上设置，而非原生代码中。原生任务失败时，错误会报告给 Airflow，然后由 Airflow 应用存根的重试策略。
• queue 的值用于将任务路由到协调器；同一个字符串必须出现在 queue_to_coordinator 中（参见 configuring-airflow-language-sdks）。

---

XCom 即 JSON 契约

XCom 值以 JSON 格式存储在 Airflow 的元数据库中，因此 Python 与任何原生语言之间的边界格式都是 JSON。所有 SDK 的 Python/JSON 对应关系都相同：

Python 类型	JSON
int	数字（整数）
float	数字（小数）
str	字符串
bool	布尔值
None	null
list	数组
dict	对象

每种语言 SDK 都会将这些 JSON 类型映射到其自身的原生类型（例如，JSON 整数会变成 Java Long）。原生类型映射详见相应语言的技能。可移植性的关键规则是：一个任务推送的值会被另一个任务作为 JSON读取，因此使用该值的一端必须预期与存储值兼容的类型。

---

与语言相关的内容（另有文档说明）

本技能有意只涵盖通用概念。以下内容因语言而异，详见各语言配套技能：

• 原生任务 API — 如何声明任务、读取连接/变量/XCom，以及推送结果（注解、接口、函数注册等）。
• 原生类型映射 — 上方 JSON 表格中的原生类型列。
• 构建与打包 — 如何编译并打包构件。
• 运行时前提条件 — Worker 上必须具备的内容（某些 SDK 需要语言运行时，例如 Java SDK 需要 JRE；Go SDK 的独立打包则不需要）。

Airflow 端的连接配置（哪个协调器运行哪个队列）在结构上通用，但各协调器的选项有所不同；详见 configuring-airflow-language-sdks。

---

与语言无关的注意事项

• ID 必须完全匹配：Python 存根函数名必须与原生任务 ID 匹配；@dag/dag_id 必须与原生 DAG ID 匹配。不匹配会导致“无 DAG”或缺少 XCom 错误。
• 两端都需要上游引用。 Python 通过传入上游调用来声明依赖；原生代码则通过 XCom 获取值。
• 在存根上设置队列和重试，不要在原生代码中设置。
• 存根主体必须为空。 AST 检查会强制执行此要求——主体中只允许有 pass、... 或文档字符串；任何实际逻辑都会被拒绝。
• 存根不支持 retry_policy（@task.stub 会引发 ValueError）。请改用 retries/retry_delay——重试策略可调用对象会在 Python 进程内运行，而在原生子进程中执行的任务无法触发它。
• 目前，语言 SDK 对资产、延迟执行及其他一些 Airflow 功能的支持有限或完全不支持。

---

各语言技能

• authoring-java-sdk-tasks：Java/Kotlin/JVM 原生 API、类型映射和日志记录。
• authoring-go-sdk-tasks：Go 原生 API — 任务注册、按参数类型进行依赖注入，以及客户端访问。
• （未来推出的各语言 SDK 都会新增自己的 authoring-<lang>-sdk-tasks 技能，并以本技能为基础。）

相关技能

• configuring-airflow-language-sdks：将队列路由到协调器并设置运行时选项。
• authoring-dags：通用 Airflow DAG 编写指南（Python 端也在此说明）。
• deploying-java-sdk-bundles：构建并交付 Java 构件。
• deploying-go-sdk-bundles：构建、打包并交付 Go 包（各语言的部署技能都采用相同结构）。