通过 GitHub 测试、同步和部署 DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本指南介绍如何创建 CI/CD 流水线，以测试、同步和部署 GitHub 代码库中的 DAG 到 Cloud Composer 环境。

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

CI/CD 流水线概览

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

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

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

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

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

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

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

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

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

目标

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

准备工作

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

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

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

  示例 CI/CD 流水线展示了示例代码库的内容。DAG 和测试存储在 dags/ 目录中，而需求文件、限制文件和 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 更改为代码库的基本分支的名称）

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

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

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

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

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

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

添加 DAG 实用程序脚本

将 DAG 实用程序脚本添加到您的代码库。此实用程序脚本将代码库的 dags/ 目录中的所有 DAG 文件都复制到临时目录中，而忽略所有非 DAG Python 文件。该脚本随后会使用 Cloud Storage 客户端库将临时目录中的所有文件都上传到 Cloud Composer 环境的存储桶中的 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 gsutil -m cp -r 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 文件，用于配置 Cloud Build 作业以同步 DAG。其中包含两个步骤：

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

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

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/**

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

在“高级配置”中，添加两个替代变量：

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

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

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

测试 CI/CD 流水线

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

运行预提交作业

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

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

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

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

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

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

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

将 DAG 添加到生产环境中

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

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

解决构建失败的问题

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

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

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

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

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

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

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

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

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

后续步骤

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