authoring-language-sdk-tasks

Authoring Language SDK Tasks (共通基盤)

Airflow の言語 SDK を使うと、DAG とスケジューリングは Python のまま、Python 以外の言語でタスクロジックを実装できます。このスキルでは、すべての言語 SDK で共通する部分について説明します。ネイティブ API、ビルドツール、ランタイムについては、各言語に専用のスキルがあります。言語別のスキルをご覧ください。

実験的機能です。 言語 SDK はプレビュー版です。API やアーティファクトの座標が変更される場合があります。

---

モデル

DAG は通常どおり Python で記述します。別の言語で実行するタスクは、専用のキューにルーティングされるスタブとして宣言します。実行時に Airflow はスタブタスクをコーディネーターに渡します。コーディネーターは、そのタスクインスタンス専用の短時間稼働するネイティブサブプロセスを起動して、コンパイル済みコードまたはネイティブコードを実行し、サブプロセスを終了します。

すべての言語 SDK に共通する事項は次のとおりです。

• タスクインスタンスごとに 1 つのサブプロセス — タスクインスタンス間でプロセス内の状態は共有されません。データは XCom または外部ストアを介して渡してください。
• DAG、スケジュール、再試行、キューへのルーティングは Python 側にあります。 ネイティブ側で実装するのはタスクロジックのみです。
• 境界を越えるデータは JSON です。 XCom の JSON 契約をご覧ください。

---

2 つの側面から成るモデル

各タスクには、互いに一致している必要がある 2 つの側面があります。

1. 通常の DAG ファイルに記述するPython スタブ — ロジックは含まず、タスク、キュー、依存関係グラフ、再試行ポリシーを宣言します。
2. Python 側の ID と一致する ID を持つネイティブ実装（Java、Go など） — 実際の処理を実行します。

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 の値は Airflow のメタデータデータベースに JSON として保存されるため、Python と任意のネイティブ言語との境界では JSON が使われます。Python と JSON の対応は、どの SDK でも共通です。

Python の型	JSON
int	数値（整数）
float	数値（小数）
str	文字列
bool	真偽値
None	null
list	配列
dict	オブジェクト

各言語 SDK は、これらの JSON 型をその言語固有のネイティブ型に対応付けます（例：JSON の整数は Java の Long になります）。ネイティブ型への対応付けは、それぞれの言語のスキルに記載されています。移植性に関する重要なルールは、あるタスクがプッシュした値は別のタスクでJSON として読み取られるため、受け取る側では保存された値と互換性のある型を想定する必要があるということです。

---

言語固有の内容（別の場所に記載）

このスキルでは、共通の概念のみを扱います。以下の内容は言語ごとに異なるため、各言語の関連スキルに記載されています。

• ネイティブタスク API — タスクの宣言方法、接続／変数／XCom の読み取り方法、結果のプッシュ方法（アノテーション、インターフェース、関数登録など）。
• ネイティブ型のマッピング — 上記の JSON テーブルのネイティブ列。
• ビルドとパッケージング — アーティファクトのコンパイルとバンドルの方法。
• ランタイムの前提条件 — ワーカー上に必要なもの（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、...、または docstring のみです。実際のロジックを含めると拒否されます。
• スタブでは retry_policy は拒否されます（@task.stub は ValueError を発生させます）。代わりに retries／retry_delay を使用してください。リトライポリシーの callable は Python をインプロセスで実行するため、ネイティブのサブプロセスで実行されるタスクでは呼び出されません。
• Assets、deferral、およびその他一部の Airflow 機能は、現在、言語 SDK でのサポートが限定的、またはサポートされていません。

---

言語別スキル

• 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 バンドルをビルド、パッケージ化して配布します（言語別のデプロイスキルも同じ構成です）。