从 GitHub 测试、同步和部署 DAG

Managed Airflow（第 3 代） | Managed Airflow（第 2 代） | Managed Airflow（传统第 1 代）

本指南介绍了如何创建 CI/CD 流水线，以便从 GitHub 代码库测试、同步 DAG 并将其部署到 Managed Airflow 环境。

如果您只想同步来自其他服务的数据，请参阅 从其他服务转移数据。

CI/CD 流水线概览

图 1. 显示流程步骤的架构图（点击可放大）

用于测试、同步和部署 DAG 的 CI/CD 流水线包含以下步骤：

1. 对 DAG 进行更改，然后将该更改推送到代码库中的开发分支 。

2. 打开针对代码库主分支的拉取请求。

3. Cloud Build 运行单元测试以检查 DAG 是否有效。

4. 您的拉取请求已获批准并会合并到您的代码库的主分支。

5. Cloud Build 将开发 Managed Airflow 环境与这些新更改同步。

6. 验证 DAG 在开发环境中是否按预期运行。

7. 如果您的 DAG 按预期运行，请将 DAG 上传到生产 Managed Airflow 环境。

目标

• 使用 Cloud Build 运行自动预提交检查。此检查会针对 DAG 运行单元测试。
• 将开发 Managed Service for Apache Airflow 环境中的 DAG 与 GitHub 代码库中的 DAG 同步。

准备工作

• 本指南假定您使用的是两个相同的 Managed Airflow 环境：开发环境和生产环境。

  就本指南而言，您仅为开发环境配置 CI/CD 流水线。请确保您使用的环境不是生产环境。

• 本指南假定您已将 DAG 及其测试存储在 GitHub 代码库中。

  示例 CI/CD 流水线展示了 示例代码库的内容。DAG 和测试存储在 dags/ 目录中，而 requirements 文件、约束文件和 Cloud Build 配置文件存储在顶层。DAG 同步实用程序及其要求位于 utils 目录中。

创建预提交检查作业和单元测试

第一个 Cloud Build 作业会运行预提交检查，该检查会针对 DAG 执行单元测试。

添加单元测试

如果尚未为 DAG 编写 单元测试，请编写这些测试。将这些测试与代码库中的 DAG 一起保存，每个测试都带有 _test 后缀。例如，example_dag.py 中 DAG 的测试文件为 example_dag_test.py。这些测试会在代码库中作为预提交检查运行。

为预提交检查创建 Cloud Build YAML 配置

在代码库中，创建一个名为 test-dags.cloudbuild.yaml 的 YAML 文件，用于为预提交检查配置 Cloud Build 作业。其中包含三个步骤：

1. 安装 DAG 所需的依赖项。
2. 安装单元测试所需的依赖项。
3. 执行 DAG 测试。

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

注意 Cloud Build 从 Docker Hub 拉取 Python 映像，您可以在每个步骤的 name 字段中指定标记。如果您使用 python 作为 entrypoint，请确保 entrypoint 与标记中指定的版本一致，否则 entrypoint 默认为 latest。

为预提交检查创建 Cloud Build 构建触发器

按照从 GitHub 构建代码库 指南中的说明，使用以下配置创建基于 GitHub 应用的触发器：

• 名称: test-dags

• 事件：拉取请求

• 来源 - 代码库：选择您的代码库

• 来源 - 基本分支：^main$（如果需要，请将 main 更改为代码库的基本分支的名称）

• 来源 - 评论控制：非必要操作

  注意 ：将“评论控制”设置为非必要操作意味着任何用户 都可以提交触发构建的拉取请求。对于较大的代码库或包含不受信任贡献者的代码库，建议使用“评论控制”。

• build 配置 - Cloud build 配置文件： /test-dags.cloudbuild.yaml（build 文件的路径）

创建 DAG 同步作业并添加 DAG 实用程序脚本

接下来，配置运行 DAG 实用程序脚本的 Cloud Build 作业。此作业中的实用程序脚本将合并到您代码库中主分支的 DAG 与 Managed Airflow 环境同步。

注意： Managed Airflow 按重复间隔解析所有 DAG。即使 DAG 自上次解析以来未发生更改，也会发生这种情况。由于存在这种行为，此实用程序会将所有 DAG 都上传到 Managed Airflow，而不仅仅是上传发生更改的 DAG。

添加 DAG 实用程序脚本

将 DAG 实用程序脚本添加到代码库。此实用程序脚本会将代码库 dags/ 目录中的所有 DAG 文件复制到临时目录，并忽略所有非 DAG Python 文件。该脚本随后会使用 Cloud Storage 客户端库将临时目录中的所有文件都上传到 Managed Airflow 环境的存储桶中的 dags/ 目录。

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gcloud storage cp --recursive on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

为同步 DAG 创建 Cloud Build YAML 配置

在代码库中，创建一个名为 add-dags-to-composer.cloudbuild.yaml 的 YAML 文件，用于为同步 DAG 配置 Cloud Build 作业。其中包含两个步骤：

1. 安装 DAG 实用程序脚本所需的依赖项。

2. 运行实用程序脚本，将代码库中的 DAG 与 Managed Airflow 环境同步。

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

注意 Cloud Build 从 Docker Hub 拉取 Python 映像，您可以在每个步骤的 name 字段中指定标记。如果您使用 python 作为 entrypoint，请确保 entrypoint 与标记中指定的版本一致，否则 entrypoint 默认为 latest。

创建 Cloud Build 构建触发器

按照从 GitHub 构建代码库 指南中的说明，使用以下配置创建基于 GitHub 应用的触发器：

• 名称: add-dags-to-composer

• 事件：推送到分支

• 来源 - 代码库：选择您的代码库

• 来源 - 基本分支：^main$（如果需要，请将 main 更改为代码库的基本分支的名称）

• 来源 - 包含的文件过滤条件 (glob)：dags/**

• build 配置 \- Cloud build 配置文件： /add-dags-to-composer.cloudbuild.yaml（build 文件的路径）

在高级配置中，添加两个 替换变量：

• _DAGS_DIRECTORY - DAG 在代码库中所在的目录。 如果您使用的是本指南中的示例代码库，则为 dags/。

• _DAGS_BUCKET - 在开发 Managed Airflow 环境中包含 dags/目录的 Cloud Storage 存储桶。省略 gs:// 前缀。例如：us-central1-example-env-1234ab56-bucket。

注意 ：如果您未从 Cloud Storage 存储桶名称中的 _DAGS_BUCKET 替换变量中移除 dags/ 后缀，DAG 同步作业 将失败并显示 NotFound 错误。发生这种情况的原因是，在 Cloud Storage 中， dags/是一个文件夹，被视为 对象名称的一部分，而不是存储桶名称的一部分。

测试 CI/CD 流水线

在本部分中，您将按照 DAG 开发流程操作以利用新创建的 Cloud Build 触发器。

运行预提交作业

创建针对主分支的拉取请求以测试 build。在页面上找到预提交检查。点击详情 ，然后选择 在 Google Cloud Build 中查看更多详情 ，以在 Google Cloud 控制台中查看 build 日志。

重要提示 ：Cloud Build 使用 requirements.txt 和 requirements-test.txt 中指定的软件包运行单元测试，其中包括 Airflow 的公开版本。您的 Managed Airflow 环境可能具有不同的 set of PyPI 软件包 和 预安装软件包。我们建议您在开发 DAG 时运行本地 Airflow 环境。

图 2.GitHub 上 Cloud Build 预提交检查状态的屏幕截图（点击可放大）

如果预提交检查失败，请参阅 解决构建失败的问题。

验证 DAG 在开发环境中是否正常运行

拉取请求获得批准后，将其合并到主分支。使用 Google Cloud 控制台 查看 build 结果。如果您有许多 Cloud Build 触发器，则可以按触发器名称 add-dags-to-composer 过滤 build。

Cloud Build 同步作业成功后，同步的 DAG 会显示在开发 Managed Airflow 环境中。您可以在其中验证 DAG 是否按预期运行。

将 DAG 添加到生产环境

DAG 按预期运行后，手动将其添加到生产环境。为此，请将 DAG 文件上传 到生产 Managed Airflow 环境的存储桶中的 dags/ 目录。

如果您的 DAG 同步作业失败或 DAG 在您的 开发 Managed Airflow 环境中未按预期运行，请参阅 解决构建失败的问题。

解决构建失败的问题

本部分介绍了如何解决常见的构建失败场景。

如果预提交检查失败，该怎么办？

在拉取请求中，点击详情 ，然后选择 在 Google Cloud Build 中查看更多详情 ，以在 Google Cloud 控制台中查看 build 日志。使用这些日志帮助您调试 DAG 问题。解决这些问题后，提交修复并推送到您的分支。预提交检查会再次运行，您可以继续使用日志作为调试工具进行迭代。

如果我的 DAG 同步作业失败，该怎么办？

使用 Google Cloud 控制台 查看 build 结果。如果您有许多 Cloud Build 触发器，则可以按触发器名称 add-dags-to-composer 过滤 build。检查 build 作业的日志并解决错误。如果您需要其他帮助来解决错误，请利用 支持渠道。

如果我的 DAG 在 Managed Airflow 环境中无法正常运行，该怎么办？

如果您的 DAG 在开发 Managed Airflow 环境中未按预期运行，请不要手动将 DAG 提升到 Managed Airflow 生产环境。请改为采用以下方法之一：

• 使用破坏 DAG 的更改还原拉取请求，以将其恢复到更改之前的状态（这也将 还原该拉取请求中的所有其他文件）。
• 创建新的拉取请求，以手动还原对损坏的 DAG 进行的更改。
• 创建新的拉取请求以修复 DAG 中的错误。

完成以下任何步骤后将触发新的预提交检查，并在合并后执行 DAG 同步作业。

后续步骤

• 运行本地 Airflow 环境
• 编写 DAG
• 安排和触发 DAG
• 测试 DAG