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
针对不受支持的Airflow运算符及复杂血缘场景的自定义OpenLineage提取器。提供两种方案:建议在自有运算符中直接添加OpenLineage方法,或为无法修改的第三方运算符创建自定义提取器。提取器在三个执行节点进行拦截:执行前获取静态血缘、成功后获取运行时输出、可选在失败后获取部分血缘。通过airflow.cfg或环境变量注册提取器...
debugging-dags
astronomer
针对失败的Airflow DAG进行系统性根因分析与修复,提供结构化调查工作流。引导完成四步诊断流程:识别故障、提取错误详情、收集上下文信息、提供可操作的修复步骤。将故障分为四类(数据、代码、基础设施、依赖),以聚焦调查并建议适当的修复方案。提供即用型CLI命令,用于日志检索、运行对比、任务清除及DAG...
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的直接消费者,构建完整的依赖树,映射从表到仪表盘再到机器学习模型的所有下游影响。按关键性(关键、高、中、低)对依赖进行分类,以优先安排利益相关者沟通和测试。生成包含风险评估、受影响...的影响报告。