迁移 Managed Airflow DAG 到…

当需要把现有 Managed Service for Apache Airflow（原 Cloud Composer）环境或本地的 DAG 代码升级到 Airflow 2.11.1（MSAA Gen 2/3）或 Airflow 3（MSAA Gen 3）时使用。

适用场景

解决思路

1. Phase 1 发现与下载：在修改前，若明确被要求，先下载现有 DAG 文件；仅在明确被要求时检查源环境以确认源版本；环境检查与文件下载的详细步骤参考 references/environment-inspection.md。
2. Phase 2.1 依赖映射（迁移到 Airflow 2.11.1 或 Airflow 3 时）：按下列版本演进追踪关键依赖，从低于 2.11.1 的 Airflow 2 迁移到 Airflow 3 时也要一并考虑。
3. Composer 2.10.0（Airflow 2.10.2）基线：Google Provider 10.26.0、SSH Provider 3.14.0、HTTP Provider 4.13.3，这是最旧且文档完整的源基线。
4. Composer 2.15.3（Airflow 2.10.5）：Google Provider 18.0.0、SSH Provider 4.1.4、HTTP Provider 5.3.4；破坏性变更包括 SSH Provider 4.0.0 移除 Hook 的 timeout、get_conn() 变为上下文管理器；HTTP Provider 5.0.0 将 SimpleHttpOperator 改为 HttpOperator；Google Provider 11.0.0 移除 BigQueryExecuteQueryOperator；12.0.0 移除旧版 Data Pipeline operators；13.0.0 移除 AutoMLBatchPredictOperator；17.0.0 移除 BigQueryCreateEmptyTableOperator、BigQueryCreateExternalTableOperator 及 Life Sciences operators；18.0.0 移除旧版 DV360 operators。
5. Composer 2.16.1（Airflow 2.10.5）：Google Provider 19.0.0、SSH Provider 4.1.6、HTTP Provider 5.5.0；Google Provider 19.0.0 移除 AutoML operators（改用 Vertex AI）。
6. Composer 2.17.0（目标 Airflow 2.11.1）：Google Provider 20.0.0、SSH Provider 5.0.0、HTTP Provider 6.0.2；SSH Provider 5.0.0 移除 sshtunnel（改用原生 tunneling）；HTTP Provider 6.0.0 使用 JSON 序列化；Google Provider 20.0.0 进行 ADLS Gen2 迁移。
7. Phase 2.2 Airflow 3 迁移：这是重大版本升级，包含 Task SDK 解耦（import 从 airflow 改为 airflow.sdk）、移除直接访问 metadata DB、Dataset 重命名为 Asset、移除 SubDAGs 和 SLAs、context 变量可用性变化；从 Airflow 2.10.2 迁移时，先应用迁移到 Airflow 2.11.1 所需的变更，再叠加 Airflow 3 迁移变更。
8. Phase 3 分析与修复：在本地工作区根目录（除非另有说明，为 ./migration_workspace）运行扫描命令。
9. 3.1.1 Dataset 调度（Airflow 2.11.0）：扫描 grep -rn "Dataset(" ./dags；修复时必须记录这些 DAG 需保持未暂停才能捕获事件，或规划手动触发以补跑。
10. 3.1.2 描述中的 HTML（Airflow 2.11.0）：扫描 grep -rn -E "doc_md.*<|doc_md.*>|description.*<|description.*>" ./dags；将 HTML 转为 Markdown，或在目标环境设置 AIRFLOW__WEBSERVER__ALLOW_RAW_HTML_DESCRIPTIONS=True。
11. 3.1.3 Teardown 任务（Airflow 2.10.5）：扫描 grep -rn "as_teardown" ./dags；确保 teardown 任务幂等。
12. 3.1.4 Pendulum 3 升级（Airflow 2.11.0）：代码扫描 grep -rn -E "pendulum\.Period|pendulum\.period" ./dags；测试扫描 grep -rn -E "\.test\(|set_test_now\(" ./tests 2>/dev/null || true；将 Period 替换为 Interval，将 period(...) 替换为 interval(...)。
13. 3.2.1 SSH Provider（SSH 4.0.0 与 5.0.0）：扫描 timeout 用 grep -rn "SSHHook" ./dags | grep "timeout"；扫描上下文管理器用 grep -rn "with SSHHook" ./dags；扫描隧道属性用 grep -rn "\.get_tunnel" ./dags；将 SSHHook 的 timeout 替换为 conn_timeout，将 with hook as conn: 替换为 with hook.get_conn() as conn:，将 get_tunnel() 作为上下文管理器使用：with hook.get_tunnel(...) as tunnel:。
14. 3.2.2 HTTP Provider（HTTP 5.0.0 与 6.0.0）：扫描 grep -rn "SimpleHttpOperator" ./dags；将 SimpleHttpOperator 替换为 HttpOperator。
15. 3.2.3 Google Provider（v11 到 v20）：BigQuery 查询扫描 grep -rn "BigQueryExecuteQueryOperator" ./dags，替换为 BigQueryInsertJobOperator（使用 configuration dict）；BigQuery 表扫描 grep -rn -E "BigQueryCreateEmptyTableOperator|BigQueryCreateExternalTableOperator" ./dags，替换为 BigQueryCreateTableOperator（使用 table_resource dict）；AutoML 扫描 grep -rn -E "AutoMLTrainModelOperator|AutoMLPredictOperator|AutoMLCreateDatasetOperator|AutoMLBatchPredictOperator" ./dags，迁移到 Vertex AI operators；Dataflow 扫描 grep -rn -E "CreateDataPipelineOperator|RunDataPipelineOperator" ./dags，替换为 DataflowCreatePipelineOperator/DataflowRunPipelineOperator；Life Sciences 扫描 grep -rn "LifeSciencesRunPipelineOperator" ./dags，迁移到 Google Cloud Batch operators（BatchCreateJobOperator）；ADLS 到 GCS 扫描 grep -rn "ADLSToGCSOperator" ./dags，确保提供 file_system_name。
16. 3.3 Airflow 3 迁移检查：迁移到 Airflow 3 时使用 references/airflow-3.md 中的说明。
17. Phase 4 部署与验证：仅在明确被要求时才执行部署与验证步骤。
18. 4.1 静态验证（迁移到 Airflow 3 时）：应用 Airflow 3 代码变更后验证语法正确性；若开发环境可用，运行静态 lint 检查 ruff check {target_dag_file} --select AIR30；在最终完成前解决所有报告的弃用警告；若 ruff 不可用，建议安装。
19. 4.2.1 获取目标 GCS Bucket 路径（仅在要求时）：运行 gcloud composer environments describe <TARGET_ENV> --location <TARGET_REGION> --format="value(config.dagGcsPrefix)"，预期输出 gs://<target-bucket-name>/dags。
20. 4.2 上传修改后的 DAG 与 Bucket 依赖（仅在明确要求时）：将本地工作区中修改后的 DAG 及备份的 bucket 依赖复制到目标 GCS bucket；若跳过了检查步骤，确保使用正确的 <target-bucket-name>；上传 DAG 用 gcloud storage cp -r ./dags/* gs://<target-bucket-name>/dags/；上传其他 bucket 依赖（如适用）用 gcloud storage cp -r ./migration_workspace/<dependency-folder> gs://<target-bucket-name>/<dependency-folder>。
21. 4.3 通过 Airflow CLI 验证 DAG（仅在明确要求上传修改后的 DAG 到目标环境时、且在上传后执行）：列出已注册 DAG 用 gcloud composer environments run <TARGET_ENV> --location <TARGET_REGION> dags list，确认迁移后的 DAG 出现在列表中；检查导入错误用 gcloud composer environments run <TARGET_ENV> --location <TARGET_REGION> dags list-import-errors，无错误时输出 No data found，有错误时列出文件路径与 traceback；注意 Airflow scheduler 解析新文件并反映到这些命令可能需要几分钟。
22. 4.4 在 Cloud Logging 中验证（仅在明确要求上传修改后的 DAG 到目标环境时、且在上传后执行）：在 GCP Cloud Logging 控制台（或通过 gcloud logging read）运行查询 resource.type="cloud_composer_environment" resource.labels.environment_name="<TARGET_ENV>" log_id("airflow-scheduler") severity>=ERROR，以检测运行时错误或导入错误。
23. 附录 本地环境验证：若想在部署到目标环境前本地验证变更，可使用 Composer Local Development CLI 工具 composer-dev，参考 references/local-development-environment.md。

适用边界 · 注意事项

• 不要用于排查与代码迁移无关的 DAG 运行失败。
• 除非明确被要求，否则不要执行环境检查、下载、部署与验证步骤。