将环境迁移到 Cloud Composer 2（来自 Airflow 2）

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

本页面介绍如何将 DAG、数据和配置从现有 Cloud Composer 1、Airflow 2 环境转移到 Cloud Composer 2、Airflow 2。

其他迁移指南

从	收件人	方法	指南
Cloud Composer 1、Airflow 2	Cloud Composer 2、Airflow 2	并排显示，使用快照	迁移指南（快照）
Cloud Composer 1、Airflow 1	Cloud Composer 2、Airflow 2	并排，使用快照	迁移指南（快照）
Cloud Composer 1、Airflow 2	Cloud Composer 2、Airflow 2	并排手动传输	本指南（手动迁移）
Cloud Composer 1、Airflow 1	Cloud Composer 2、Airflow 2	并排手动转移	手动迁移指南
Airflow 1	Airflow 2	并排手动转移	手动迁移指南

准备工作

• Cloud Composer 支持从 Cloud Composer 1 并行迁移到 Cloud Composer 2。无法从 Cloud Composer 1 就地升级到 Cloud Composer 2。
• 查看 Cloud Composer 1 和 Cloud Composer 2 之间的差异列表。

第 1 步：获取配置替换、自定义 PyPI 软件包和环境变量的列表

控制台

获取 Cloud Composer 1 环境的配置替换、自定义 PyPI 软件包和环境变量的列表：

1. 前往 Google Cloud 控制台中的环境页面：

   转到“环境”

2. 选择 Cloud Composer 1 环境。

3. 在环境变量标签页上查看环境变量。

4. 在 Airflow 配置替换标签页上查看配置替换。

5. 在 PyPI 软件包标签页中查看自定义 PyPI 软件包。

gcloud

如需获取环境变量列表，请运行以下命令：

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.envVariables)"

如需获取环境的 Airflow 配置替换列表，请运行以下命令：

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.airflowConfigOverrides)"

如需获取自定义 PyPI 软件包的列表，请运行以下命令：

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.pypiPackages)"

替换：

• 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
• 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。

Terraform

跳过此步骤。Cloud Composer 1 环境的配置已列出了您的环境的配置替换、自定义 PyPI 软件包和环境变量。

第 2 步：创建 Cloud Composer 2 环境

在此步骤中，创建一个 Cloud Composer 2 环境。您可以从符合预期资源需求的环境预设开始，然后再进一步扩缩和优化环境。

控制台

创建 Cloud Composer 2 环境并指定配置替换和环境变量。

或者，您可以在创建环境后替换 Airflow 配置和环境变量。

gcloud

创建 Cloud Composer 2 环境并指定配置替换和环境变量。

或者，您可以在创建环境后替换 Airflow 配置和环境变量。

Terraform

根据 Cloud Composer 1 环境的配置创建 Cloud Composer 2 环境：

1. 复制 Cloud Composer 1 环境的配置。
2. 更改环境的名称。

3. 使用 google-beta 提供方：

   resource "google_composer_environment" "example_environment_composer_2" {
     provider = google-beta
     # ...
   }
   

4. 在 config.software_config 块中指定 Cloud Composer 2 映像：

   software_config {
     image_version = "composer-2.9.7-airflow-2.9.3"
     # ...
   }
   

5. 如果尚未指定配置替换和环境变量，请指定。

6. 在 config.software_config.pypi_packages 块中指定自定义 PyPI 软件包：

   software_config {
   
     # ...
   
     pypi_packages = {
       numpy = ""
       scipy = ">=1.1.0"
     }
   
   }
   

第 3 步：将 PyPI 软件包安装到 Cloud Composer 2 环境中

注意：如果您的环境不使用自定义 PyPI 软件包，请跳过此步骤。自定义 PyPI 软件包是除了环境中的预装软件包之外您安装的软件包。

创建 Cloud Composer 2 环境后，请向其中安装自定义 PyPI 软件包。

控制台

1. 前往 Google Cloud 控制台中的环境页面：

   转到“环境”

2. 选择 Cloud Composer 2 环境。

3. 转到 PyPI 软件包标签页，然后点击修改。

4. 从 Cloud Composer 1 环境复制 PyPI 软件包要求。点击保存，然后等待您的环境更新。

gcloud

1. 使用自定义 PyPI 软件包列表创建一个 requirements.txt 文件：

     numpy
     scipy>=1.1.0
   

2. 更新环境并将 requirements.txt 文件传递给 --update-pypi-packages-from-file 命令：

   gcloud composer environments update COMPOSER_2_ENV \
     --location COMPOSER_2_LOCATION  \
     --update-pypi-packages-from-file requirements.txt
   

   替换：

   • 将 COMPOSER_2_ENV 替换为 Cloud Composer 2 环境的名称。
   • 将 COMPOSER_2_LOCATION 替换为 Cloud Composer 2 环境所在的区域。

Terraform

跳过此步骤。创建环境时，您已经安装了自定义 PyPI 软件包。

第 4 步：转移变量和池

注意：如果您的环境不使用 default_pool 以外的 Airflow 变量或池，请跳过此步骤。

Airflow 支持将变量和池导出到 JSON 文件。然后，您可以将这些文件导入 Cloud Composer 2 环境。

此步骤中使用的 Airflow CLI 命令适用于 Airflow 工作器中的本地文件。如需上传或下载文件，请使用环境的 Cloud Storage 存储桶中的 /data 文件夹。此文件夹会同步到 Airflow 工作器中的 /home/airflow/gcs/data/ 目录。在 Airflow CLI 命令的参数 FILEPATH 中指定 /home/airflow/gcs/data/。

gcloud

1. 从 Cloud Composer 1 环境中导出变量：

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
        variables export -- /home/airflow/gcs/data/variables.json
   

   替换：

   • 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
   • 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。

2. 从 Cloud Composer 1 环境中导出池：

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
        pools export -- /home/airflow/gcs/data/pools.json
   

   替换：

   • 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
   • 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。

3. 获取 Cloud Composer 2 环境的存储桶 URI。

   1. 运行以下命令：

      gcloud composer environments describe COMPOSER_2_ENV \
          --location COMPOSER_2_LOCATION \
           --format="value(config.dagGcsPrefix)"
      

      替换：

      • 将 COMPOSER_2_ENV 替换为 Cloud Composer 2 环境的名称。
      • 将 COMPOSER_2_LOCATION 替换为环境所在的区域。

   2. 在输出中，移除 /dags 文件夹。结果是 Cloud Composer 2 环境存储桶的 URI。

      例如，将 gs://us-central1-example-916807e1-bucket/dags 更改为 gs://us-central1-example-916807e1-bucket。

4. 将包含变量和池的 JSON 文件转移到 Cloud Composer 2 环境：

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=variables.json
   

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=pools.json
   

   替换：

   • 将 COMPOSER_2_BUCKET 替换为在上一步获取的 Cloud Composer 2 环境存储桶的 URI。
   • 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
   • 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。

5. 将变量和池导入 Cloud Composer 2：

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       variables import \
       -- /home/airflow/gcs/data/variables.json
   

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       pools import \
       -- /home/airflow/gcs/data/pools.json
   

6. 检查已导入的变量和池：

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       variables list
   

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       pools list
   

7. 从存储分区中移除 JSON 文件：

   gcloud composer environments storage data delete \
       variables.json \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

   gcloud composer environments storage data delete \
       pools.json \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

   gcloud composer environments storage data delete \
       variables.json \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

   gcloud composer environments storage data delete \
       pools.json \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

第 5 步：从 Cloud Composer 1 环境的存储桶转移其他数据

从 Cloud Composer 1 环境的存储桶转移插件和其他数据。

gcloud

1. 将插件转移到 Cloud Composer 2 环境。为此，请将插件从 Cloud Composer 1 环境的存储桶导出到 Cloud Composer 2 环境存储桶的 /plugins 文件夹中：

   gcloud composer environments storage plugins export \
       --destination=COMPOSER_2_BUCKET/plugins \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

2. 检查 /plugins 文件夹是否已成功导入：

   gcloud composer environments storage plugins list \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

3. 将 /data 文件夹从 Cloud Composer 1 环境导出到 Airflow 2 环境：

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

4. 检查 /data 文件夹是否已成功导入：

   gcloud composer environments storage data list \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

第 6 步：转移连接

注意：Cloud Composer 环境有许多已为环境配置的默认连接。请勿转移这些连接，因为 Cloud Composer 2 环境已有这些连接。如果您的环境仅使用默认连接，请跳过此步骤。

此步骤说明如何通过手动创建连接来转移连接。

gcloud

1. 如需获取 Cloud Composer 1 环境中的连接列表，请运行以下命令：

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       connections list
   

2. 如需在 Cloud Composer 2 环境中创建新连接，请通过 gcloud 运行 connections Airflow CLI 命令。例如：

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       connections add \
       -- --conn-host postgres.example.com \
       --conn-port 5432 \
       --conn-type postgres \
       --conn-login example_user \
       --conn-password example_password \
       --conn-description "Example connection" \
       example_connection
   

第 7 步：转移用户账号

此步骤说明如何通过手动创建用户来转移用户。

注意：如需查看用户，您的账号必须具有 Admin 角色。

Airflow 界面

1. 如需查看 Cloud Composer 1 环境中的用户列表，请执行以下操作：

   1. 打开 Cloud Composer 1 环境的 Airflow 网页界面。

   2. 转到安全性 > 列出用户。

2. 如需在 Cloud Composer 2 环境中创建用户，请执行以下操作：

   1. 打开 Cloud Composer 2 环境的 Airflow 网页界面。

   2. 转到安全性 > 列出用户。

   3. 点击添加新记录。

gcloud

1. 如需查看 Cloud Composer 1 环境中的用户列表，请通过 gcloud 运行 users list Airflow CLI 命令：

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       users list
   

   替换：

   • 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
   • 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。

2. 如需在 Cloud Composer 2 环境中创建新的用户账号，请运行 users create Airflow CLI 命令 至 gcloud。例如：

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       users create \
       -- --username example_username \
       --firstname Example-Name \
       --lastname Example-Surname \
       --email [e-mail] \
       --use-random-password \
       --role Op
   

   替换：

   • 将 COMPOSER_2_ENV 替换为 Cloud Composer 2 环境的名称。
   • 将 COMPOSER_2_LOCATION 替换为 Cloud Composer 2 环境所在的区域。
   • Cloud Composer 1 环境中的所有用户配置参数及其值，包括用户的角色。

转移用户账号的其他方法

作为替代方案，您可以使用 users export 和 users import Airflow CLI 命令。

1. 将用户账号导出到环境的存储桶 /data 文件夹中的文件：

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       users export -- /home/airflow/gcs/data/users.json
   

2. 将此文件导出到 Cloud Composer 2 环境的存储桶：

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=users.json
   

3. 将用户账号从此文件导入 Cloud Composer 2 环境：

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       users import \
       -- /home/airflow/gcs/data/users.json
   

4. 删除两个环境中的 JSON 文件：

   gcloud composer environments storage data delete \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION  \
       users.json
   

   gcloud composer environments storage data delete \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION  \
       users.json
   

替换：

• 将 COMPOSER_1_ENV 替换为 Cloud Composer 1 环境的名称。
• 将 COMPOSER_2_ENV 替换为 Cloud Composer 2 环境的名称。
• 将 COMPOSER_1_LOCATION 替换为 Cloud Composer 1 环境所在的区域。
• 将 COMPOSER_2_LOCATION 替换为 Cloud Composer 2 环境所在的区域。
• 将 COMPOSER_2_BUCKET 替换为在上一步获取的 Cloud Composer 2 环境存储桶的 URI。

第 8 步：将 DAG 转移到 Cloud Composer 2 环境

警告：此步骤说明了在环境之间转移 DAG 时如何避免与重复、缺失、错误及其他无效 DAG 运行相关的潜在问题。

在环境之间转移 DAG 时，可能会出现以下问题：

• 如果两个环境中都启用了（未暂停）某个 DAG，则每个环境都会安排自己的 DAG 副本。这可能会导致针对相同的数据和执行时间重复运行 DAG。

• 由于 DAG 同步，Airflow 从 DAG 中指定的开始日期开始安排额外的 DAG 运行。之所以发生这种情况，是因为新的 Airflow 实例不考虑从 Cloud Composer 1 环境运行 DAG 的历史记录。这可能会导致从指定的开始日期开始安排大量 DAG 运行。

防止重复的 DAG 运行

在 Cloud Composer 2 环境中或者在 Airflow 2 环境中为 dags_are_paused_at_creation 选项添加 Airflow 配置选项替换。进行此项更改后，默认情况下，所有新 DAG 都会暂停。

部分	键	值
core	dags_are_paused_at_creation	True

注意：在将所有 DAG 转移到 Cloud Composer 2 环境后，您可以移除此替换值。

防止额外运行或缺失的 DAG 运行

为避免执行日期出现间隔和重叠，请在 Cloud Composer 2 中停用同步。通过这种方式，您在将 DAG 上传到 Cloud Composer 2 环境后，Airflow 不会安排已在 Cloud Composer 1 环境中进行的 DAG 运行作业。为 catchup_by_default 选项添加 Airflow 配置选项替换：

部分	键	值
scheduler	catchup_by_default	False

注意：请勿移除此替换值。如果您这样做，您的环境会同步 DAG 运行。这些 DAG 运行可能与已在 Cloud Composer 1 环境中执行的运行重叠。

将 DAG 转移到 Cloud Composer 2 环境

如需将 DAG 转移到 Cloud Composer 2 环境，请执行以下操作：

1. 将 DAG 从 Cloud Composer 1 环境上传到 Cloud Composer 2 环境。跳过 airflow_monitoring.py DAG。

2. 由于配置替换，DAG 在 Cloud Composer 2 环境中被暂停，因此未安排任何 DAG 运行。

3. 在 Airflow 网页界面中，转到 DAG 并查看报告的 DAG 语法错误。

4. 当您计划转移 DAG 时，请执行以下操作：

   1. 在 Cloud Composer 1 环境中暂停 DAG。

   2. 在 Cloud Composer 2 环境中取消暂停 DAG。

   3. 检查新的 DAG 运行是否安排在正确的时间。

   4. 等待 DAG 在 Cloud Composer 2 环境中运行并检查运行是否成功。如果 DAG 运行成功，请不要在 Cloud Composer 1 环境中取消暂停该 DAG。如果这样做，则 DAG 会在 Cloud Composer 1 环境中以相同的时间和日期运行。

5. 如果特定 DAG 运行失败，请尝试对该 DAG 进行问题排查，直到它在 Cloud Composer 2 中成功运行为止。

   如果需要，您可以随时回退到 DAG 的 Cloud Composer 1 版本，并从 Cloud Composer 1 环境执行在 Cloud Composer 2 中失败的 DAG 运行：

   1. 在 Cloud Composer 2 环境中暂停 DAG。

   2. 在 Cloud Composer 1 环境中取消暂停 DAG。这安排了 DAG 在 Cloud Composer 1 环境中暂停时的 DAG 运行情况。

第 9 步：监控 Cloud Composer 2 环境

将所有 DAG 和配置转移到 Cloud Composer 2 环境后，请监控其是否存在潜在问题、DAG 运行失败以及环境整体运行状况。如果 Cloud Composer 2 环境在足够长的时间内正常运行，请考虑删除 Cloud Composer 1 环境。

后续步骤

• DAG 问题排查
• 排查环境创建问题
• 排查环境更新问题
• 使用反向移植软件包