authoring-go-sdk-tasks

Go SDK タスクの作成

Airflow Go SDK は、Go 向けの言語 SDK モデルを実装しています。DAG は Python に保持し、各タスクはコンパイルされた Go 関数として登録されます。この関数は、単一のネイティブ実行ファイルであるバンドルに含まれます。このスキルでは、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 スタブと概念）、deploying-go-sdk-bundles（バンドルのビルド、パッケージ化、配布）、configuring-airflow-language-sdks（キューを Go コーディネーターにルーティング）。

---

おさらい：Python 側

Go タスクには、ロジックを持たない Python スタブを対応させます。スタブは、タスク、キュー、依存関係グラフを宣言します。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 を参照してください。

---

バンドルのエントリーポイント

バンドルは bundlev1.BundleProvider を実装し、そのバージョンを報告して DAG とタスクを登録します。main は 1 行で済み、bundlev1server.Serve がバンドルを 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 スタブと一致させられない場合（非公開関数、名前を変更した関数、または再利用する関数）は、AddTaskWithName("<task_id>", fn) を使用してください。RegisterDags がタスク ID の唯一の情報源です。バンドルのマニフェスト（パッカーとコーディネーターが使用）は、手書きするのではなく、バンドルを実行して生成します。

---

タスク関数：パラメーター型による依存性注入

タスクは通常の 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 が nil でなければ、タスクの return_value XCom としてプッシュされます。error が nil でなければ、タスクは失敗し（スタブのリトライポリシーが適用されます）。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（スキーマ）、Extra map[string]any、および GetURI() を公開します。見つからない場合は、センチネル値 sdk.VariableNotFound、sdk.ConnectionNotFound、sdk.XComNotFound を返します。

上流タスクの結果を読み取るには、GetXCom を明示的に呼び出します。必要な dag_id／run_id／task_id は、以下の実行時コンテキストから取得してください。

---

実行時コンテキスト

タスクインスタンスとその DAG 実行に関するメタデータを読み取るには、sdk.TIRunContext パラメーターを宣言します。これは 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 の場合はパニックします）。

---

Go 固有の注意点

• ID は Python スタブと一致させる必要があります（AddDag の dag_id、登録された関数名の task_id）。また、スタブの queue= が Go コーディネーターにルーティングされていないと、タスクは配信されません。
• RegisterDags が正式な定義です。 マニフェストを手書きしないでください。パッカーは RegisterDags を実行してマニフェストを生成します。
• 必要最小限のクライアントインターフェースを要求してください（sdk.Client より sdk.VariableClient）。意図が明確になり、フェイクも使いやすくなります。
• error の戻り値が nil 以外の場合、タスクは失敗し、スタブの再試行が適用されます。リカバリーされたパニックも失敗として扱われます。
• 言語に依存しない注意点（タスクインスタンスごとに 1 プロセス、スタブでキューと再試行を設定）については、authoring-language-sdk-tasks を参照してください。

---

関連スキル

• authoring-language-sdk-tasks：Python スタブの共通パターンと概念（最初にお読みください）。
• deploying-go-sdk-bundles：go tool airflow-go-pack でバンドルをビルドしてパックし、コーディネーター用にデプロイします。
• configuring-airflow-language-sdks：キューを Go コーディネーター（ExecutableCoordinator）にルーティングします。
• authoring-dags：Airflow DAG の一般的な作成方法。