authoring-go-sdk-tasks

编写 Go SDK 任务

Airflow Go SDK 为 Go 实现了语言 SDK 模型：DAG 保留在 Python 中，每个任务都是一个已编译的 Go 函数，并注册在一个 bundle（单个原生可执行文件）中。本技能介绍 Go 专属的原生 API。共享模型（Python @task.stub 模式、ID 匹配、XCom 作为 JSON 的约定）位于 authoring-language-sdk-tasks；如果你刚接触语言 SDK，请先阅读该文档。

实验性功能。 Go SDK 正在积极开发中，尚未达到生产就绪状态。模块路径为 github.com/apache/airflow/go-sdk（Go 1.24+）。API 可能会变更。

相关技能： authoring-language-sdk-tasks（共享的 Python stub 和概念）、deploying-go-sdk-bundles（构建、打包和分发 bundle）、configuring-airflow-language-sdks（将队列路由到 Go 协调器）。

---

回顾：Python 端

Go 任务需要配套一个不包含逻辑的 Python stub；它声明任务、队列和依赖关系图。ID 必须与 Go 注册项完全匹配，queue= 会将任务路由到 Go 运行时。完整规则见 authoring-language-sdk-tasks；最简示例如下：

from airflow.sdk import dag, task


@task.stub(queue="golang")
def extract(): ...


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


@dag()
def simple_dag():
    extract() >> transform()


simple_dag()

queue 的值（此处为 "golang"）是一个任意标签，必须与路由到 Go 协调器的队列（queue_to_coordinator）相匹配。参见 configuring-airflow-language-sdks。

---

bundle 入口点

bundle 实现 bundlev1.BundleProvider：报告其版本，并注册 DAG 和任务。main 只有一行；bundlev1server.Serve 会自动将 bundle 接入 Airflow 运行时。

package main

import (
	"log"

	v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
	"github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
)

type myBundle struct{}

var _ v1.BundleProvider = (*myBundle)(nil)

func (m *myBundle) GetBundleVersion() v1.BundleInfo {
	return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
}

func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
	simpleDag := dagbag.AddDag("simple_dag")      // dag_id must match the Python @dag name
	simpleDag.AddTask(extract)                    // task_id is the function name; must match the stub
	simpleDag.AddTaskWithName("transform", transform) // or set the task_id explicitly
	return nil
}

func main() {
	if err := bundlev1server.Serve(&myBundle{}); err != nil {
		log.Fatal(err)
	}
}

AddTask(fn) 会根据 Go 函数名生成 task_id；如果函数名无法与 Python stub 匹配（例如函数未导出、已重命名或重复使用），请使用 AddTaskWithName("<task_id>", fn)。RegisterDags 是任务身份的唯一事实来源：bundle 的清单（供打包器和协调器使用）由运行它生成，绝不能手动编写。

---

任务函数：按参数类型注入依赖

任务是普通的 Go 函数。运行时会检查其签名并按类型注入参数；只声明你需要的参数。

参数类型	注入的值
context.Context	用于取消操作的任务上下文。始终可用。
sdk.TIRunContext	更丰富的上下文（嵌入 context.Context），提供 TaskInstance() 和 DagRun()。参见运行时上下文。
*slog.Logger	接入 Airflow 任务日志的日志记录器。
sdk.Client	完整的 Airflow 模型访问：Variables、Connections、XComs。
sdk.VariableClient / sdk.ConnectionClient / sdk.XComClient	sdk.Client 的精简接口。优先选择所需的最精简接口；这样既能清晰表达意图，也很容易在测试中模拟。

可选的返回签名为 (result, error)：非空的 result 会作为任务的 return_value XCom 推送；非空的 error 会使任务失败（从而触发 stub 的重试策略）。也可以只返回 error，或不返回任何值。

func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any, error) {
	conn, err := client.GetConnection(ctx, "test_http")
	if err != nil {
		return nil, err
	}
	log.Info("connected", "host", conn.Host)
	return map[string]any{"go_version": runtime.Version()}, nil
}

func transform(ctx sdk.TIRunContext, client sdk.VariableClient) error {
	val, err := client.GetVariable(ctx, "my_variable")
	if err != nil {
		return err // VariableNotFound (a sentinel error) if absent
	}
	_ = val
	return nil
}

sdk.Client 接口

调用	返回值	说明
GetVariable(ctx, key)	(string, error)	如果不存在，则返回 VariableNotFound。
UnmarshalJSONVariable(ctx, key, &ptr)	error	将 JSON 变量解码为结构体/指针。
GetConnection(ctx, connID)	(Connection, error)	如果不存在，则返回 ConnectionNotFound。
GetXCom(ctx, dagID, runID, taskID, mapIndex, key, value)	(any, error)	仅当键不存在时返回 XComNotFound；已存储的 null 会返回 (nil, nil)。
PushXCom(ctx, ti, key, value)	error	很少需要；返回的值会自动推送。

Connection 提供 ID、Type、Host、Port（int）、Login *string、Password *string（未设置时为 nil，与空值不同）、Path（schema）、Extra map[string]any，以及 GetURI()。未找到时会返回哨兵值 sdk.VariableNotFound、sdk.ConnectionNotFound、sdk.XComNotFound。

若要读取上游任务的结果，请显式调用 GetXCom，并从运行时上下文（如下）中获取所需的 dag_id/run_id/task_id。

---

运行时上下文

声明一个 sdk.TIRunContext 参数，以读取有关任务实例及其 DAG 运行的元数据。它是一个嵌入了 context.Context 的接口，因此可在任何需要 context.Context 的地方使用。

func extract(ctx sdk.TIRunContext, log *slog.Logger) error {
	ti, dagRun := ctx.TaskInstance(), ctx.DagRun()
	log.Info("running",
		"task_id", ti.TaskID,
		"run_id", dagRun.RunID,
		"logical_date", dagRun.LogicalDate)
	return nil
}

• TaskInstance()：DagID、RunID、TaskID、MapIndex *int（未映射时为 nil）、TryNumber。
• DagRun()：DagID、RunID，以及 *time.Time 类型的时间戳 LogicalDate、DataIntervalStart、DataIntervalEnd（未发送时为 nil）。

访问器会在函数体运行前，根据任务启动详情填充。由于 TIRunContext 嵌入了 context.Context，可将其直接传给客户端调用和取消检查（ctx.Done()）；默认将其声明为上下文参数。在测试中，使用 sdk.NewTIRunContext(ctx, ti, dagRun) 构建参数（如果 ctx 为 nil，则会触发 panic）。

---

Go 特有的注意事项

• ID 必须与 Python stub 一致（AddDag 中的 dag_id、已注册函数名中的 task_id），且 stub 的 queue= 必须路由到 Go 协调器，否则任务永远不会被投递。
• RegisterDags 是权威来源。 不要手写清单；打包器会通过运行 RegisterDags 生成清单。
• 只请求所需的最窄客户端接口（需要 sdk.VariableClient 时优先于 sdk.Client），以便明确意图并简化伪造实现。
• 返回非 nil 的 error 会使任务失败，并应用 stub 中的重试设置；恢复的 panic 也会导致失败。
• 语言无关的注意事项（每个任务实例使用一个进程、在 stub 上设置 queue 和重试）请参阅 authoring-language-sdk-tasks。

---

相关技能

• authoring-language-sdk-tasks：Python stub 的通用模式和概念（请先阅读）。
• deploying-go-sdk-bundles：使用 go tool airflow-go-pack 构建并打包 bundle，然后将其部署到协调器。
• configuring-airflow-language-sdks：将队列路由到 Go 协调器（ExecutableCoordinator）。
• authoring-dags：Airflow DAG 编写通用指南。