airflow-new-sdk

作成者: astronomer

Airflow向けの新しい言語SDK(AIP-108)を実装するためのガイド。新しいプログラミング言語のサポートを追加したいコントリビューター向けのスキルです。

npx skills add https://github.com/astronomer/airflow --skill airflow-new-sdk

Implementing a new language SDK for Airflow

Start here

Read contributing-docs/30_new_language_sdk.rst first. It is the authoritative contributor guide for this topic — coordinator base class choices, wire protocol spec, bundle footer format, and testing requirements. Everything in this skill builds on top of it, not alongside it.


Repository layout

Every new SDK needs two things. The coordinator (Python) goes here:

task-sdk/src/airflow/sdk/coordinators/<language>/
    __init__.py        # re-export + module docstring
    coordinator.py     # subclass of SubprocessCoordinator or BaseCoordinator
task-sdk/tests/coordinators/<language>/
    test_coordinator.py
task-sdk/tests/integration/coordinators/<language>/
    test_integration.py   # requires Breeze

The language SDK itself lives in a top-level <language>-sdk/ directory (like java-sdk/ and go-sdk/). For native-executable languages using ExecutableCoordinator, no coordinator code is needed at all.


Choosing the right base class — quick guide

Does the runtime compile to a self-contained native executable?
  YES → Use ExecutableCoordinator (zero Python to write).
        Append an AFBNDL01 footer with a packer tool (see go-sdk reference).
  NO  →
    Does it start via a single shell command (node, ruby, dotnet, …)?
      YES → Subclass SubprocessCoordinator.
            Implement _build_execute_task_command only (see 30_new_language_sdk.rst).
      NO  →
        Subclass BaseCoordinator and implement execute_task from scratch.
        (Rare: gRPC daemons, shared memory, persistent processes.)

The full rationale, method signature, and socket lifecycle for each path are in 30_new_language_sdk.rst. Read that section before writing any code.


Reference implementations to study

What to studyWhere
SubprocessCoordinator base classtask-sdk/src/airflow/sdk/coordinators/_subprocess.py
Java coordinator (SubprocessCoordinator subclass)task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
ExecutableCoordinator (native bundles)task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py
Wire protocol in Kotlinjava-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/
Wire protocol in Gogo-sdk/pkg/execution/
AFBNDL01 footer (Go reference)go-sdk/internal/bundlefooter/, task-sdk/docs/executable-bundle-spec.rst
All message types and field specstask-sdk/src/airflow/sdk/execution_time/schema/schema.json

The Java and Go implementations are the two production reference points. When implementing a new SDK, read whichever matches the target language's runtime model (JVM/interpreted → Java; native/compiled → Go).


Logging

The Logging section of 30_new_language_sdk.rst is the spec: the --logs JSON record format, the level names, and the AIRFLOW__LOGGING__* environment variables. Read it first. A few language-neutral details it leaves out:

  • Level values. Levels follow Python's logging scale, so thresholds line up with the rest of Airflow: CRITICAL=50, ERROR=40, WARNING=30, INFO=20, DEBUG=10, NOTSET=0.
  • Parsing NAMESPACE_LEVELS. Split the value on [\s,]+, then split each item on = into (logger_name, level_name). Emit a record only when its level is >= the matching logger_name threshold, or the global threshold when no per-logger entry matches.
  • Don't drop late logs. Connect the --logs socket early and keep it open until the --comm channel has finished; otherwise records emitted during teardown can be lost.
  • Extra config keys. The runtime can't read Airflow's config, so if your SDK needs [logging] settings beyond the two above, propagate them as environment variables from your coordinator's start, the same way.

For how a given language wires its native logging frameworks into this channel, read that SDK's source (e.g. java-sdk/) rather than reproducing it here.

E2E test suite

Create two files mirroring java_sdk_tests/ or go_sdk_tests/:

airflow-e2e-tests/tests/airflow_e2e_tests/<language>_sdk_tests/
    __init__.py
    test_<language>_sdk_dag.py

The test file should:

  • Trigger the SDK's example Dag (the one added to <language>-sdk/dags/ or equivalent) via AirflowClient.trigger_dag.
  • Wait for the run to finish with AirflowClient.wait_for_dag_run.
  • Assert that each SDK task instance reached "success".
  • Assert at least one XCom value — confirms the full round-trip from task return value through the supervisor to the XCom store.
  • Assert structured logs where the SDK emits them (see the Go suite for an example of log-content assertions).

Run locally with:

E2E_TEST_MODE=<language>_sdk uv run --project airflow-e2e-tests pytest \
    tests/airflow_e2e_tests/<language>_sdk_tests/ -xvs

The Java (java_sdk_tests/test_java_sdk_dag.py) and Go (go_sdk_tests/test_go_sdk_dag.py) suites are the reference implementations.


PR checklist (items not covered by 30_new_language_sdk.rst)

The 30_new_language_sdk.rst guide covers coordinator placement, wire protocol implementation, and testing. These additional items belong in the same PR:

  1. task-sdk/src/airflow/sdk/coordinators/<language>/__init__.py — short module docstring and __all__ re-export of the coordinator class.
  2. airflow-core/docs/authoring-and-scheduling/language-sdks/<language>.rst — user-facing doc following the structure of java.rst or go.rst.
  3. airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst — add the new doc to the toctree.
  4. airflow-core/newsfragments/<PR>.feature.rst — a new language is always user-visible; add a newsfragment.
  5. CI wiring — check dev/breeze/src/airflow_breeze/utils/selective_checks.py to confirm <language>-sdk/ changes trigger the right test group. Add if missing.
  6. E2E tests — add a <language>_sdk_tests/ suite under airflow-e2e-tests/tests/airflow_e2e_tests/. See below.

astronomerのその他のスキル

airflow-state-store
astronomer
Persists task and asset state across retries and DAG runs using Airflow 3.3's AIP-103 key/value stores (`task_state_store`, `asset_state_store`) and the…
creating-openlineage-extractors
astronomer
カスタムOpenLineage抽出器。未対応のAirflowオペレーターや複雑な系列シナリオ向け。2つのアプローチ:所有するオペレーターに直接OpenLineageメソッドを追加する方法(推奨)、または変更できないサードパーティ製オペレーター用にカスタム抽出器を作成する方法。抽出器は3つの時点でオペレーターの実行をインターセプトします:静的な系列のための実行前、実行時に決定される出力のための成功後、およびオプションで部分的な系列のための失敗後。抽出器はairflow.cfgまたは環境変数経由で登録...
debugging-dags
astronomer
失敗したAirflow DAGに対する体系的な根本原因分析と修正、構造化された調査ワークフローを提供。4段階の診断プロセス(障害の特定、エラー詳細の抽出、コンテキスト情報の収集、実行可能な修正手順の提示)をガイド。障害を4つのタイプ(データ、コード、インフラストラクチャ、依存関係)に分類し、調査を集中させ適切な修正を提案。ログ取得、実行比較、タスククリア、DAG...のための即時使用可能なCLIコマンドを提供。
delegating-to-otto
astronomer
Drives Astronomer's Otto agent (`astro otto`) as a delegated sub-agent for Airflow, dbt, and data-engineering work. Use when the user explicitly asks to "use…
deploying-airflow
astronomer
AirflowのDAGやプロジェクトをデプロイします。ユーザーがコードのデプロイ、DAGのプッシュ、CI/CDの設定、本番環境へのデプロイ、またはデプロイ戦略について質問した場合に使用します…
deploying-go-sdk-bundles
astronomer
コンパイルされたAirflow Go SDKバンドルをビルド、パッケージ化、デプロイし、ExecutableCoordinatorが実行できるようにします。ユーザーがGoタスクバンドルをコンパイルしたい場合、または…と尋ねた場合に使用します。
testing-dags
astronomer
Airflow DAGに対する反復的なテスト・デバッグ・修正サイクルと、包括的な障害診断を提供します。af runs trigger-wait <dag_id> でDAGを実行し完了を待機します。事前チェックは不要です。失敗時はaf runs diagnoseで包括的な障害サマリーを取得し、af tasks logsで特定タスクのエラー詳細を確認できます。カスタム設定、タイムアウト、リトライ試行に対応。成功、失敗、タイムアウトの各シナリオを明確な応答解釈で処理します。迅速な検証が可能です...
tracing-downstream-lineage
astronomer
テーブルやDAGを変更する前に、下流のデータ系列を追跡して変更影響を評価します。ソースコード検索、ビュー依存関係、BIツール接続を通じて、対象テーブルまたはDAGの直接的な消費者を特定します。テーブルからダッシュボード、MLモデルに至るまで、すべての下流影響をマッピングする完全な依存関係ツリーを構築します。依存関係を重要度(クリティカル、高、中、低)で分類し、ステークホルダーへの連絡とテストの優先順位付けを行います。リスク評価と影響を受けるものを含む影響レポートを生成します。