SuperAPI AI 工具集

airflow-new-sdk

编程与开发

查看项目

更新于 10-15 19:58

技能介绍

用途:指导贡献者为 Airflow 新增一门编程语言的 SDK(AIP-108)。应先阅读 contributing-docs/30_new_language_sdk.rst,它是该主题的权威指南(基类选择、wire protocol 规范、bundle footer 格式与测试要求)。

主要能力与结构:每个新 SDK 需要两部分。协调器(Python)放在 task-sdk/src/airflow/sdk/coordinators/<language>/,含 init.py 与 coordinator.py,并有 task-sdk/tests/coordinators/<language>/ 与 task-sdk/tests/integration/coordinators/<language>/ 测试;语言 SDK 本体放在顶层 <language>-sdk/ 目录(如 java-sdk/、go-sdk/)。

基类选择:运行时能编译为自包含原生可执行文件时用 ExecutableCoordinator(无需写 Python),并用打包工具追加 AFBNDL01 footer;通过单一 shell 命令启动时(node、ruby、dotnet 等)继承 SubprocessCoordinator,只需实现 _build_execute_task_command;否则继承 BaseCoordinator 从头实现 execute_task(少见,如 gRPC 守护进程、共享内存、常驻进程)。

参考实现:Java 与 Go 是两套生产参考。JVM/解释型参照 Java,原生/编译型参照 Go;相关路径包括 _subprocess.py、java/coordinator.py、executable/coordinator.py、java-sdk 的 Kotlin wire protocol、go-sdk/pkg/execution/、go-sdk/internal/bundlefooter/,消息类型见 execution_time/schema/schema.json。

日志:遵循 30_new_language_sdk.rst 的 --logs JSON 记录格式、级别名与 AIRFLOW__LOGGING__* 环境变量。级别沿用 Python 日志标度(CRITICAL=50、ERROR=40、WARNING=30、INFO=20、DEBUG=10、NOTSET=0);NAMESPACE_LEVELS 以 [\s,]+ 拆分再按 = 拆成 (logger_name, level_name),仅当级别不低于匹配阈值时输出;应尽早连接 --logs socket 并保持到 --comm 结束,以免丢失收尾日志;超出配置的 [logging] 项需由协调器启动时以环境变量传入。

测试与 PR:参照 java_sdk_tests/ 或 go_sdk_tests/ 创建 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 等待完成,断言任务成功、至少一个 XCom 值和结构化日志。本地运行:E2E_TEST_MODE=<language>_sdk uv run --project airflow-e2e-tests pytest tests/airflow_e2e_tests/<language>_sdk_tests/ -xvs。PR 还需:init.py 模块文档与 all 导出、airflow-core/docs 的 <language>.rst 及 index toctree、newsfragments 的 .feature.rst、CI 选择性检查 wiring 与 E2E 套件。

限制:资料未提供具体语言的完整实现代码、依赖版本或耗时估计,实施前须以 30_new_language_sdk.rst 与 Java/Go 源码为准。