managed-airflow-migrations

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Managed Service for Apache Airflow (formerly Cloud Composer) Migration Guide

托管式Apache Airflow服务（前身为Cloud Composer）迁移指南

This skill guides you through the process of adjusting Airflow DAGs from an existing Managed Service for Apache Airflow (formerly Cloud Composer) environment (or available locally) to make them compatible with Airflow 2.11.1 (MSAA Gen 2 or 3) or Airflow 3 (MSAA Gen 3).

本技能将引导您调整现有托管式Apache Airflow服务（前身为Cloud Composer）环境（或本地可用环境）中的Airflow DAG，使其兼容Airflow 2.11.1（MSAA第2代或第3代）或Airflow 3（MSAA第3代）。

Phase 1: Discovery & Download

阶段1：发现与下载

Before making any changes, download the existing DAG files if explicitly requested. Inspect the source environment to confirm source version only if explicitly requested. For detailed instructions about environment inspection and downloading files check references/environment-inspection.md.

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

Phase 2: Target Version & Dependency Mapping

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

2.1 Airflow 2.11.1+ Dependency Mapping

2.1 Airflow 2.11.1+依赖映射

If migrating to Airflow 2.11.1 (MSAA Gen 2) or Airflow 3, use the list below to trace the version progression of key dependencies. The list covers changes needed to get to Airflow 2.11.1. Take them into account when migrating from Airflow 2 (earlier than 2.11.1) to Airflow 3.

如果迁移至Airflow 2.11.1（MSAA第2代）或Airflow 3，请使用以下列表追踪关键依赖的版本演进。该列表涵盖了升级至Airflow 2.11.1所需的变更。从Airflow 2（早于2.11.1版本）迁移至Airflow 3时，请将这些变更纳入考量。

Composer 2.10.0 (Airflow 2.10.2)

Composer 2.10.0（Airflow 2.10.2）

Google Provider:
```
10.26.0
```
SSH Provider:
```
3.14.0
```
HTTP Provider:
```
4.13.3
```
Breaking Changes: Baseline for oldest fully documented source.

Google Provider：
```
10.26.0
```
SSH Provider：
```
3.14.0
```
HTTP Provider：
```
4.13.3
```
破坏性变更：最早的完整文档化源版本基线。

Composer 2.15.3 (Airflow 2.10.5)

Composer 2.15.3（Airflow 2.10.5）

Google Provider:
```
18.0.0
```
SSH Provider:
```
4.1.4
```
HTTP Provider:
```
5.3.4
```
Breaking Changes:
- SSH Provider 4.0.0: Hook
```
timeout
```
  removed;
```
get_conn()
```
  context manager.
- HTTP Provider 5.0.0:
```
SimpleHttpOperator
```
  ->
```
HttpOperator
```
  .
- Google Provider 11.0.0:
```
BigQueryExecuteQueryOperator
```
  removed.
- Google Provider 12.0.0: Legacy Data Pipeline operators removed.
- Google Provider 13.0.0:
```
AutoMLBatchPredictOperator
```
  removed.
- Google Provider 17.0.0:
```
BigQueryCreateEmptyTableOperator
```
  and
```
BigQueryCreateExternalTableOperator
```
  removed; Life Sciences operators removed.
- Google Provider 18.0.0: Legacy DV360 operators removed.

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)

Composer 2.16.1（Airflow 2.10.5）

Google Provider:
```
19.0.0
```
SSH Provider:
```
4.1.6
```
HTTP Provider:
```
5.5.0
```
Breaking Changes: Google Provider 19.0.0: AutoML operators removed (use Vertex AI).

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 (Target Airflow 2.11.1)

Composer 2.17.0（目标Airflow 2.11.1）

Google Provider: 20.0.0
SSH Provider: 5.0.0
HTTP Provider: 6.0.2
Breaking Changes:
- SSH Provider 5.0.0:
```
sshtunnel
```
  removed (native tunneling).
- HTTP Provider 6.0.0: JSON serialization.
- Google Provider 20.0.0: ADLS Gen2 migration.

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 Migration

2.2 Airflow 3迁移

If migrating to Airflow 3 (MSAA Gen 3), note that this is a major version upgrade with significant changes, including:

Decoupled Task SDK (imports change from
```
airflow
```
to
```
airflow.sdk
```
).
Removal of direct metadata DB access.
Renaming of
```
Dataset
```
to
```
Asset
```
.
Removal of SubDAGs and SLAs.
Changes to context variables availability.

Take into account all applicable changes within Airflow 2 (e.g. when migrating from Airflow 2.10.2, apply changes needed to move to Airflow 2.11.1 and Airflow 3 migration changes on top of that).

如果迁移至Airflow 3（MSAA第3代），请注意这是一个重大版本升级，包含诸多显著变更，例如：

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

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

Phase 3: Analysis & Remediation (Scanning Downloaded Files)

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

Run the scan commands from the root of your local workspace (

./migration_workspace

unless indicated otherwise).

从本地工作区根目录（除非另有说明，否则为

./migration_workspace

）运行扫描命令。

3.1 Airflow 2.11.1 Core & Dependency checks

3.1 Airflow 2.11.1核心与依赖检查

Use these scans if migrating to Airflow 2.11.1+ (intermediate step when migrating to Airflow 3).

若迁移至Airflow 2.11.1+（迁移至Airflow 3的中间步骤），请使用以下扫描。

3.1.1 Dataset Scheduling (Airflow 2.11.0)

3.1.1 Dataset调度（Airflow 2.11.0）

Change: DAGs scheduled on datasets only trigger if events occur while the DAG is unpaused.
Scan Command:
```
grep -rn "Dataset(" ./dags
```
Remediation: You MUST document that these DAGs must remain unpaused to catch events, or plan manual triggers for catch-up.

变更：基于Dataset调度的DAG仅在DAG未暂停时发生事件才会触发。
扫描命令：
```
grep -rn "Dataset(" ./dags
```
修复方案：您必须记录这些DAG需保持未暂停状态才能捕获事件，或计划手动触发以补跑任务。

3.1.2 HTML in Descriptions (Airflow 2.11.0)

3.1.2 描述中的HTML（Airflow 2.11.0）

Change: Raw HTML in DAG docs / params is escaped by default.

Scan Command:

bash

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

Remediation: Convert HTML to Markdown, or set
```
AIRFLOW__WEBSERVER__ALLOW_RAW_HTML_DESCRIPTIONS=True
```
in target.

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

扫描命令：

bash

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

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

3.1.3 Teardown Tasks (Airflow 2.10.5)

3.1.3 Teardown任务（Airflow 2.10.5）

Change: Teardowns always run when a DAG is marked failed.
Scan Command:
```
grep -rn "as_teardown" ./dags
```
Remediation: Ensure teardown tasks are idempotent.

变更：当DAG标记为失败时，Teardown任务总会运行。
扫描命令：
```
grep -rn "as_teardown" ./dags
```
修复方案：确保Teardown任务具有幂等性。

3.1.4 Pendulum 3 Upgrade (Airflow 2.11.0)

3.1.4 Pendulum 3升级（Airflow 2.11.0）

Change:
```
Period
```
renamed to
```
Interval
```
, testing helpers removed.

Scan Command (Code):

bash

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

Scan Command (Tests):

bash

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

Remediation: Replace
```
Period
```
with
```
Interval
```
, and
```
period(...)
```
with
```
interval(...)
```
.

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

代码扫描命令：

bash

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

测试扫描命令：

bash

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

修复方案：将
```
Period
```
替换为
```
Interval
```
，将
```
period(...)
```
替换为
```
interval(...)
```
。

3.2 Path A: Airflow 2.11.1 Provider Package Scan

3.2 路径A：Airflow 2.11.1 Provider包扫描

3.2.1 SSH Provider (SSH 4.0.0 & 5.0.0)

3.2.1 SSH Provider（SSH 4.0.0 & 5.0.0）

Scan Command (Timeout):

grep -rn "SSHHook" ./dags | grep "timeout"

Scan Command (Context Manager):
```
grep -rn "with SSHHook" ./dags
```
Scan Command (Tunnel Attributes):
```
grep -rn "\.get_tunnel" ./dags
```

Remediation:

Replace
```
timeout
```
with
```
conn_timeout
```
in
```
SSHHook
```
.

Replace

with hook as conn:

with

with hook.get_conn() as conn:

Use

get_tunnel()

as context manager:

with hook.get_tunnel(...) as tunnel:

超时扫描命令：

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)

3.2.2 HTTP Provider（HTTP 5.0.0 & 6.0.0）

Scan Command:
```
grep -rn "SimpleHttpOperator" ./dags
```
Remediation: Replace
```
SimpleHttpOperator
```
with
```
HttpOperator
```
.

扫描命令：
```
grep -rn "SimpleHttpOperator" ./dags
```
修复方案：将
```
SimpleHttpOperator
```
替换为
```
HttpOperator
```
。

3.2.3 Google Provider (v11 to v20)

3.2.3 Google Provider（v11至v20）

Scan Command (BigQuery query):

bash

grep -rn "BigQueryExecuteQueryOperator" ./dags

Remediation: Replace with
```
BigQueryInsertJobOperator
```
(use
```
configuration
```
dict).

Scan Command (BigQuery table):

bash

grep -rn -E "BigQueryCreateEmptyTableOperator|BigQueryCreateExternalTableOperator" ./dags

Remediation: Replace with

BigQueryCreateTableOperator

(use

table_resource

dict).

Scan Command (AutoML):

bash

grep -rn -E "AutoMLTrainModelOperator|AutoMLPredictOperator|AutoMLCreateDatasetOperator|AutoMLBatchPredictOperator" ./dags

Remediation: Migrate to Vertex AI operators.

Scan Command (Dataflow):

bash

grep -rn -E "CreateDataPipelineOperator|RunDataPipelineOperator" ./dags

Remediation: Replace with

DataflowCreatePipelineOperator

DataflowRunPipelineOperator

Scan Command (Life Sciences):
bash
```
grep -rn "LifeSciencesRunPipelineOperator" ./dags`
```
- Remediation: Migrate to Google Cloud Batch operators (
```
BatchCreateJobOperator
```
  ).
Scan Command (ADLS to GCS):
```
grep -rn "ADLSToGCSOperator" ./dags
```
- Remediation: Ensure
```
file_system_name
```
  is provided.

BigQuery查询扫描命令：

bash

grep -rn "BigQueryExecuteQueryOperator" ./dags

修复方案：替换为
```
BigQueryInsertJobOperator
```
（使用
```
configuration
```
字典）。

BigQuery表扫描命令：

bash

grep -rn -E "BigQueryCreateEmptyTableOperator|BigQueryCreateExternalTableOperator" ./dags

修复方案：替换为
```
BigQueryCreateTableOperator
```
（使用
```
table_resource
```
字典）。

AutoML扫描命令：

bash

grep -rn -E "AutoMLTrainModelOperator|AutoMLPredictOperator|AutoMLCreateDatasetOperator|AutoMLBatchPredictOperator" ./dags

修复方案：迁移至Vertex AI算子。

Dataflow扫描命令：

bash

grep -rn -E "CreateDataPipelineOperator|RunDataPipelineOperator" ./dags

修复方案：替换为

DataflowCreatePipelineOperator

DataflowRunPipelineOperator

。

Life Sciences扫描命令：
bash
```
grep -rn "LifeSciencesRunPipelineOperator" ./dags`
```
- 修复方案：迁移至Google Cloud Batch算子（
```
BatchCreateJobOperator
```
  ）。
ADLS转GCS扫描命令：
```
grep -rn "ADLSToGCSOperator" ./dags
```
- 修复方案：确保提供
```
file_system_name
```
  参数。

3.3 Airflow 3 Migration checks

3.3 Airflow 3迁移检查

Use instructions from references/airflow-3.md when migrating to Airflow 3.

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

Phase 4: Deployment & Verification

阶段4：部署与验证

Perform deployment and verification steps only if explicitly requested to do so.

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

4.1 Static Verification (when migrating to Airflow 3)

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

After applying code changes for Airflow 3, verify syntax correctness. If available in the development environment, run static lint checks:

bash

ruff check {target_dag_file} --select AIR30

Resolve any reported deprecation warnings before finalization. If ruff is not available, recommend installing one.

完成Airflow 3的代码变更后，验证语法正确性。若开发环境中可用，运行静态代码检查：

bash

ruff check {target_dag_file} --select AIR30

在最终确认前解决所有报告的弃用警告。若未安装ruff，建议安装该工具。

4.2 Deployment to MSAA

4.2 部署至MSAA

4.2.1 Get Target GCS Bucket Path (only when requested)

4.2.1 获取目标GCS存储桶路径（仅在要求时执行）

bash

gcloud composer environments describe <TARGET_ENV> \
    --location <TARGET_REGION> \
    --format="value(config.dagGcsPrefix)"

Expected Output:

gs://<target-bucket-name>/dags

bash

gcloud composer environments describe <TARGET_ENV> \
    --location <TARGET_REGION> \
    --format="value(config.dagGcsPrefix)"

预期输出：

gs://<target-bucket-name>/dags

4.2 Upload Modified DAGs and Bucket Dependencies (Only when requested)

4.2 上传修改后的DAG及存储桶依赖（仅在要求时执行）

Perform this step only if explicitly requested to do so. Copy the modified DAGs and any backed-up bucket dependencies from your local workspace to the target GCS bucket. If you skipped the inspection step, ensure you have the correct
<target-bucket-name>
.

Upload DAGs:

bash

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

Upload Other Bucket Dependencies (If applicable):

bash

gcloud storage cp -r ./migration_workspace/<dependency-folder> gs://<target-bucket-name>/<dependency-folder>

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

上传DAG：

bash

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

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

bash

gcloud storage cp -r ./migration_workspace/<dependency-folder> gs://<target-bucket-name>/<dependency-folder>

4.3 Verify DAGs via Airflow CLI

4.3 通过Airflow CLI验证DAG

Perform this step only if explicitly requested to upload modified DAGS to a target environment (and after uploading).

You can verify that your DAGs have been successfully uploaded, parsed, and registered by the Airflow scheduler in the target environment using the Airflow CLI.

List Registered DAGs: Run the following command to list all DAGs registered in the target environment. Verify that your migrated DAGs appear in this list.
bash
```
gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list
```
Check for Import Errors: If some DAGs are missing from the list, or to ensure there are no parsing issues, check for import errors:
bash
```
gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list-import-errors
```
Expected Output:
- If there are no errors, the command will output
```
No data found
```
  .
- If there are errors, it will list the file path and the traceback of the error.

Note: It may take a couple of minutes for the Airflow scheduler to parse the new files and for changes to reflect in these commands.

仅在明确要求将修改后的DAG上传至目标环境时（且上传完成后）执行此步骤。

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

列出已注册的DAG：运行以下命令列出目标环境中所有已注册的DAG。验证您迁移的DAG是否出现在列表中。
bash
```
gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list
```
检查导入错误：若部分DAG未出现在列表中，或需确保无解析问题，请检查导入错误：
bash
```
gcloud composer environments run <TARGET_ENV> \
    --location <TARGET_REGION> \
    dags list-import-errors
```
预期输出：
- 若无错误，命令将输出
```
No data found
```
  。
- 若存在错误，将列出文件路径和错误回溯信息。

注意：Airflow调度器解析新文件并使变更在这些命令中生效可能需要几分钟时间。

4.4 Verify in Cloud Logging

4.4 通过Cloud Logging验证

Perform this step only if explicitly requested to upload modified DAGS to a target environment (and after uploading). Monitor Cloud Logging for the target environment to detect any runtime errors or import errors.

Run the following query in the GCP Cloud Logging Console (or via

gcloud logging read

query

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

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

在GCP Cloud Logging控制台（或通过

gcloud logging read

）运行以下查询：

query

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

Appendix: Local Environment Verification

附录：本地环境验证

If you want to verify your changes locally before deploying to the target environment, you can use the Composer Local Development CLI tool (

composer-dev

). Use references/local-development-environment.md as a reference for interactions with local development environments.

若您希望在部署到目标环境前在本地验证变更，可以使用Composer本地开发CLI工具（

composer-dev

）。与本地开发环境交互的参考文档请查看references/local-development-environment.md。