SuperAPI AI 工具集

airflow-new-sdk

编程与开发

查看项目

更新于 10-15 19:57

技能介绍

用途:为 Apache Airflow 实现一门全新编程语言的 SDK(AIP-108),面向希望新增语言支持的贡献者。

主要能力与要点:

  1. 先阅读权威指南 contributing-docs/30_new_language_sdk.rst,其中包含协调器基类选择、线协议规范、bundle footer 格式与测试要求。
  2. 仓库布局:Python 协调器放在 task-sdk/src/airflow/sdk/coordinators/<language>/(含 init.py 与 coordinator.py),测试放在 task-sdk/tests/coordinators/<language>/ 与 task-sdk/tests/integration/coordinators/<language>/(需 Breeze)。语言 SDK 本身位于顶层 <language>-sdk/ 目录(如 java-sdk/、go-sdk/)。
  3. 基类选择:运行时编译为自包含原生可执行文件时用 ExecutableCoordinator(无需写 Python,需用打包工具追加 AFBNDL01 footer);通过单一 shell 命令启动(node、ruby、dotnet 等)时继承 SubprocessCoordinator,仅实现 _build_execute_task_command;否则继承 BaseCoordinator 从头实现 execute_task(少见,如 gRPC 守护进程、共享内存、常驻进程)。
  4. 参考实现:Java 与 Go 是两套生产参考。JVM/解释型参考 Java,原生/编译型参考 Go。相关文件包括 _subprocess.py、java/coordinator.py、executable/coordinator.py、Kotlin 与 Go 的 execution 目录、schema.json 等。
  5. 日志:遵循 30_new_language_sdk.rst 的 Logging 章节,--logs JSON 记录格式、级别名称与 AIRFLOW__LOGGING__* 环境变量。级别沿用 Python logging 数值(CRITICAL=50、ERROR=40、WARNING=30、INFO=20、DEBUG=10、NOTSET=0);解析 NAMESPACE_LEVELS 时按 [\s,]+ 拆分,再按 = 拆成 logger_name 与 level_name,仅当级别不低于匹配阈值时输出;尽早连接 --logs socket 并保持到 --comm 结束,避免丢失收尾日志;额外配置键需由协调器启动时以环境变量传递。
  6. E2E 测试:在 airflow-e2e-tests/tests/airflow_e2e_tests/<language>sdk_tests/ 下创建 init.py 与 test<language>_sdk_dag.py,通过 AirflowClient.trigger_dag 触发示例 DAG,用 wait_for_dag_run 等待完成,断言任务实例为 success、至少一个 XCom 值、以及结构化日志。本地运行命令:E2E_TEST_MODE=<language>_sdk uv run --project airflow-e2e-tests pytest tests/airflow_e2e_tests/<language>_sdk_tests/ -xvs
  7. PR 清单补充项:协调器 init.py 的模块 docstring 与 all 重导出;airflow-core/docs/authoring-and-scheduling/language-sdks/<language>.rst 用户文档并加入 index.rst toctree;airflow-core/newsfragments/<PR>.feature.rst;检查 dev/breeze/src/airflow_breeze/utils/selective_checks.py 的 CI 触发;补充 E2E 测试套件。

限制:资料未提供具体收费、价格、平台或效果信息;实现细节以官方指南与参考实现为准。