mcpField
M
managed-airflow-migrations

managed-airflow-migrations

github.com/google数据与数据库21.6k4.4k 次安装今天更新
代码仓库

关于 managed-airflow-migrations

帮助将 Managed Service for Apache Airflow 的 DAG 调整为兼容 Airflow 2.11.1 或 Airflow 3 的迁移技能，面向维护现有 Airflow 工作流的团队。它涵盖依赖项变化、代码扫描与修复，并将部署和验证限定在明确提出请求时。

主要功能

  1. 01梳理目标版本的依赖变化
  2. 02扫描 DAG 中的兼容性问题
  3. 03提供常见代码迁移建议
  4. 04支持 Airflow 3 迁移检查
  5. 05按请求执行部署与验证

适用场景

  1. 01升级 DAG 至 Airflow 2.11.1
  2. 02将 Airflow 2 工作流迁移至 Airflow 3
  3. 03部署前检查 DAG 兼容问题

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/google/skills --skill managed-airflow-migrations

安装后在 Claude Code 中输入/managed-airflow-migrations确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

references/
airflow-3.md
attribution.md
environment-inspection.md
local-development-environment.md

详细介绍

Managed Service for Apache Airflow（原名 Cloud Composer）迁移指南

本技能将指导你调整现有 Managed Service for Apache Airflow（原名 Cloud Composer）环境中的 DAG（或本地可用的 DAG），使其兼容 Airflow 2.11.1（MSAA Gen 2 或 3）或 Airflow 3（MSAA Gen 3）。

归属声明

必须阅读：执行任何 gcloud、REST API 或 MCP 命令之前，必须先阅读 使用归属声明。对于直接在终端中执行的命令（绝不用于生成的脚本、Terraform 或用户代码），请在 gcloud 前加上 CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:managed-airflow-migrations)" 并在 User-Agent 标头中追加 agent-skills/1.0 (skill:managed-airflow-migrations)。

阶段 1：发现与下载

进行任何更改之前，仅在明确要求时下载现有 DAG 文件。仅在明确要求时检查源环境以确认源版本。有关环境检查和文件下载的详细说明，请参阅 references/environment-inspection.md。

阶段 2：目标版本与依赖映射

2.1 Airflow 2.11.1+ 依赖映射

如果要迁移到 Airflow 2.11.1（MSAA Gen 2）或 Airflow 3，请使用下方列表追踪关键依赖项的版本演进。该列表涵盖升级到 Airflow 2.11.1 所需的变更。从 Airflow 2（早于 2.11.1）迁移到 Airflow 3 时，请一并考虑这些变更。

Composer 2.10.0（Airflow 2.10.2）
  • Google Provider：10.26.0
  • SSH Provider：3.14.0
  • HTTP Provider：4.13.3
  • 破坏性变更：有完整文档说明的最早源版本基线。
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。
    • **Google Provider 12.0.0：**移除了旧版 Data Pipeline 运算符。
    • **Google Provider 13.0.0：**移除了 AutoMLBatchPredictOperator。
    • **Google Provider 17.0.0：**移除了 BigQueryCreateEmptyTableOperator 和 BigQueryCreateExternalTableOperator；移除了 Life Sciences 运算符。
    • **Google Provider 18.0.0：**移除了旧版 DV360 运算符。
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 运算符（改用 Vertex AI）。
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（改用原生隧道）。
    • **HTTP Provider 6.0.0：**JSON 序列化。
    • **Google Provider 20.0.0：**ADLS Gen2 迁移。
2.2 Airflow 3 迁移

如果要迁移到 Airflow 3（MSAA Gen 3），请注意这属于主版本升级，会带来重大变更，包括：

  • 解耦 Task SDK（导入路径从 airflow 改为 airflow.sdk）。
  • 移除直接访问元数据数据库的方式。
  • 将 Dataset 重命名为 Asset。
  • 移除 SubDAG 和 SLA。
  • 上下文变量的可用性发生变化。

请一并考虑 Airflow 2 中所有适用的变更（例如，从 Airflow 2.10.2 迁移时，先应用升级到 Airflow 2.11.1 所需的变更，再叠加 Airflow 3 迁移变更）。

阶段 3：分析与修复（扫描已下载的文件）

从本地工作区根目录运行扫描命令 （除非另有说明，否则为 ./migration_workspace）。

3.1 Airflow 2.11.1 核心与依赖项检查

如果要迁移到 Airflow 2.11.1+（迁移到 Airflow 3 的中间步骤），请使用这些扫描。

3.1.1 数据集调度（Airflow 2.11.0）
  • 变更： 仅当 DAG 未暂停时发生事件，按数据集调度的 DAG 才会触发。
  • 扫描命令： grep -rn "Dataset(" ./dags
  • 修复： 必须记录这些 DAG 必须保持未暂停状态才能捕获事件，或计划通过手动触发来补跑。
3.1.2 描述中的 HTML（Airflow 2.11.0）

  • 变更： DAG 文档 / 参数中的原始 HTML 默认会被转义。

  • 扫描命令：

    grep -rn -E "doc_md.*<|doc_md.*>|description.*<|description.*>" ./dags

  • 修复： 将 HTML 转换为 Markdown，或在目标环境中设置 AIRFLOW__WEBSERVER__ALLOW_RAW_HTML_DESCRIPTIONS=True。

3.1.3 拆除任务（Airflow 2.10.5）
  • 变更： 将 DAG 标记为失败时，拆除任务始终会运行。
  • 扫描命令： grep -rn "as_teardown" ./dags
  • 修复： 确保拆除任务具备幂等性。
3.1.4 Pendulum 3 升级（Airflow 2.11.0）

  • 变更： Period 已重命名为 Interval，测试辅助工具已移除。

  • 扫描命令（代码）：

    grep -rn -E "pendulum\.Period|pendulum\.period" ./dags

  • 扫描命令（测试）：

    grep -rn -E "\.test\(|set_test_now\(" ./tests 2>/dev/null || true

  • 修复： 将 Period 替换为 Interval，并将 period(...) 替换为 interval(...)。

3.2 路径 A：Airflow 2.11.1 Provider 软件包扫描
3.2.1 SSH Provider（SSH 4.0.0 和 5.0.0）
  • 扫描命令（超时）： 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:。
3.2.2 HTTP Provider（HTTP 5.0.0 和 6.0.0）
  • 扫描命令： grep -rn "SimpleHttpOperator" ./dags
  • 修复： 将 SimpleHttpOperator 替换为 HttpOperator。
3.2.3 Google 提供商（v11 到 v20）

  • 扫描命令（BigQuery 查询）：

    grep -rn "BigQueryExecuteQueryOperator" ./dags
    • 修复方法： 替换为 BigQueryInsertJobOperator（使用 configuration 字典）。

  • 扫描命令（BigQuery 表）：

    grep -rn -E "BigQueryCreateEmptyTableOperator|BigQueryCreateExternalTableOperator" ./dags
    • 修复方法： 替换为 BigQueryCreateTableOperator（使用 table_resource 字典）。

  • 扫描命令（AutoML）：

    grep -rn -E "AutoMLTrainModelOperator|AutoMLPredictOperator|AutoMLCreateDatasetOperator|AutoMLBatchPredictOperator" ./dags
    • 修复方法： 迁移到 Vertex AI 运算符。

  • 扫描命令（Dataflow）：

    grep -rn -E "CreateDataPipelineOperator|RunDataPipelineOperator" ./dags
    • 修复方法： 替换为 DataflowCreatePipelineOperator/DataflowRunPipelineOperator。

  • 扫描命令（Life Sciences）：

    grep -rn "LifeSciencesRunPipelineOperator" ./dags`
    • 修复方法： 迁移到 Google Cloud Batch 运算符 （BatchCreateJobOperator）。

  • 扫描命令（ADLS 到 GCS）： grep -rn "ADLSToGCSOperator" ./dags

    • 修复方法： 确保提供了 file_system_name。
3.3 Airflow 3 迁移检查

迁移到 Airflow 3 时，请使用 references/airflow-3.md 中的说明。

阶段 4：部署与验证

仅在明确要求时执行部署和验证步骤。

4.1 静态验证（迁移到 Airflow 3 时）

应用 Airflow 3 的代码更改后，请验证语法是否正确。如果开发环境中有相应工具，请运行静态 lint 检查：

ruff check {target_dag_file} --select AIR30

在最终定稿前，解决所有报告的弃用警告。如果没有 ruff，建议安装一个。

4.2 部署到 MSAA
4.2.1 获取目标 GCS 存储桶路径（仅在请求时）
gcloud composer environments describe <TARGET_ENV> \
    --location <TARGET_REGION> \
    --format="value(config.dagGcsPrefix)"

预期输出： gs://<target-bucket-name>/dags

4.2 上传修改后的 DAG 和存储桶依赖项（仅在请求时）

仅在明确要求时执行此步骤。 将修改后的 DAG 和所有已备份的存储桶依赖项从本地工作区复制到目标 GCS 存储桶。如果跳过了检查步骤，请确保使用了正确的 <target-bucket-name>。

  1. 上传 DAG：

    gcloud storage cp -r ./dags/* gs://<target-bucket-name>/dags/

  2. 上传其他存储桶依赖项（如适用）：

    gcloud storage cp -r ./migration_workspace/<dependency-folder> gs://<target-bucket-name>/<dependency-folder>
4.3 通过 Airflow CLI 验证 DAG

仅当明确要求将修改后的 DAG 上传到目标环境时，才执行此步骤（并在上传后执行）。

你可以使用 Airflow CLI 验证 DAG 是否已成功上传、解析并由目标环境中的 Airflow 调度器注册。

  1. 列出已注册的 DAG： 运行以下命令，列出目标环境中所有已注册的 DAG。确认迁移后的 DAG 出现在此列表中。

    gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list

  2. 检查导入错误： 如果列表中缺少某些 DAG，或要确保没有解析问题，请检查导入错误：

    gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list-import-errors

    预期输出：

    • 如果没有错误，命令将输出 No data found。
    • 如果有错误，将列出文件路径和错误的回溯信息。

注意：Airflow 调度器可能需要几分钟才能解析新文件，并让更改反映在这些命令的结果中。

4.4 在 Cloud Logging 中验证

仅当明确要求将修改后的 DAG 上传到目标环境时，才执行此步骤（并在上传后执行）。 监控目标环境的 Cloud Logging，以检测任何运行时错误或导入错误。

在 GCP Cloud Logging Console 中运行以下查询（或通过 gcloud logging read 运行）：

resource.type="cloud_composer_environment"
resource.labels.environment_name="<TARGET_ENV>"
log_id("airflow-scheduler")
severity>=ERROR

附录：本地环境验证

如果你想在部署到目标环境之前在本地验证更改，可以使用 Composer Local Development CLI 工具 （composer-dev）。请参考 references/local-development-environment.md， 了解如何与本地开发环境交互。

评论

支持 Markdown