在 Managed Airflow（第 2 代）中运行 Apache Airflow DAG

在 Managed Airflow（第 2 代）中运行 Apache Airflow DAG

Managed Airflow（第 3 代） | Managed Airflow（第 2 代） | Managed Airflow（旧版第 1 代）

本快速入门指南介绍了如何创建 Managed Service for Apache Airflow 环境，并在 Managed Airflow（第 2 代）中运行 Apache Airflow DAG。

• 如果您是 Airflow 新手，请参阅 Apache Airflow 文档中的 Airflow 概念教程，详细了解 Airflow 概念、对象及其 用法。

• 如果您想改用 Google Cloud CLI，请参阅 在 Managed Service for Apache Airflow 中运行 Apache Airflow DAG (Google Cloud CLI)。

• 如果您想使用 Terraform 创建环境，请参阅 创建环境 (Terraform)。

准备工作

1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

2. Verify that billing is enabled for your Google Cloud project.

3. 启用 Managed Airflow API。

   启用 API 所需的角色

   如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您 创建了项目，则可能已通过 所有者角色 (roles/owner) 拥有此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获取此权限。 了解如何授予角色。

   启用 API

4. 如需获得完成本快速入门所需的权限，请让您的管理员为您授予项目的以下 IAM 角色：

   • 分配 IAM 角色和权限： Project IAM Admin (roles/resourcemanager.projectIamAdmin)
   • 为 Managed Airflow 环境创建服务帐号： Create Service Accounts (roles/iam.serviceAccountCreator)
   • 查看、创建和管理 Managed Airflow 环境：
     • Environment and Storage Object Administrator (roles/composer.environmentAndStorageObjectAdmin)
     • Service Account User (roles/iam.serviceAccountUser)
   • 查看日志： Logs Viewer (roles/logging.viewer)

   如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

   您也可以通过自定义 角色或其他预定义 角色来获取所需的权限。

创建环境的服务帐号

创建环境时，您需要指定服务帐号。此服务 账号称为“环境的服务账号”。您的环境使用此服务帐号执行大多数操作。

您的环境的服务帐号不是用户账号。服务帐号是由应用或虚拟机 (VM) 实例（而非单个用户）使用的特殊账号。

如需为您的环境创建服务帐号，请执行以下操作：

1. 按照 Identity and Access Management 文档中的说明创建新的服务账号。

2. 按照 Identity and Access Management 文档中的说明向其授予角色。所需的角色是 Composer Worker (composer.worker)。

创建环境

1. 在 Google Cloud 控制台中，前往创建环境页面。

   转到“创建环境”

1. 如果这是您项目中的第一个环境，则会显示向 Cloud Composer 服务账号授予所需权限 部分。

   您需要将 Managed Airflow 服务代理账号作为新正文添加到环境的服务帐号中，并向其授予 Cloud Composer v2 API Service Agent Extension 角色。

   确认您为环境使用了预期的服务帐号，然后点击授予 。

2. 在名称 字段中，输入 example-environment。

3. 在位置 下拉列表中，为 Managed Airflow 环境选择一个区域。本指南使用 us-central1 区域。

4. 对于其他环境配置选项，请使用提供的默认值。

5. 点击创建 ，然后等待环境创建完成。

   注意： Managed Airflow 大约需要 25 分钟 才能创建您的环境。

6. 完成后，相应环境名称旁边会显示一个绿色对勾标记。

创建 DAG 文件

Airflow DAG 是您要安排和运行的有序任务的集合 。DAG 在标准 Python 文件中定义。

本指南使用 quickstart.py 文件中定义的示例 Airflow DAG。 此文件中的 Python 代码会执行以下操作：

1. 创建一个 DAG composer_sample_dag。此 DAG 每天运行一次。
2. 执行一项任务 print_dag_run_conf。该任务使用 bash 运算符输出 DAG 运行的配置。

在本地机器上保存 quickstart.py 文件的副本：

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

将 DAG 文件上传到环境的存储桶

每个 Managed Airflow 环境都有一个关联的 Cloud Storage 存储桶。Managed Airflow 中的 Airflow 仅会调度此存储桶中 /dags 文件夹内的 DAG。

如需安排 DAG，请将 quickstart.py 从本地机器上传到您的环境的 /dags 文件夹：

1. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

2. 在环境列表中，点击环境的名称 example-environment。环境详情 页面会打开。

3. 点击打开 DAG 文件夹 。存储分区详情 页面会打开。

4. 点击上传文件 ，然后选择 quickstart.py 的副本。

5. 如需上传该文件，请点击打开 。

查看 DAG

上传 DAG 文件后，Airflow 会执行以下操作：

1. 解析您上传的 DAG 文件。Airflow 可能需要几分钟才能使用 DAG。
2. 将 DAG 添加到可用 DAG 列表中。
3. 根据您在 DAG 文件中提供的时间表执行 DAG。

在 DAG 界面中查看 DAG，以检查 DAG 是否已处理完毕且无错误，以及是否可在 Airflow 中使用。DAG 界面是 Managed Airflow 界面，用于在 控制台中查看 Google Cloud DAG 信息。Managed Airflow 还提供 对 Airflow 界面的访问权限，后者是原生 Airflow 网页 界面。

1. 等待大约 5 分钟，以便 Airflow 处理您之前上传的 DAG 文件，并完成第一次 DAG 运行（稍后会对此进行说明）。

2. 在 Google Cloud 控制台中，前往环境页面。

   转到“环境”

3. 在环境列表中，点击环境的名称 example-environment。环境详情 页面会打开。

4. 前往 DAG 标签页。

5. 检查 DAG 列表中是否存在 composer_quickstart DAG。

   

   图 1.DAG 列表显示 composer_quickstart DAG（点击可放大）

   注意： airflow_monitoring DAG 由 Managed Airflow 用于 监控环境的运行状况。

查看 DAG 运行详情

DAG 的单次执行称为“DAG 运行”。 Airflow 会立即为示例 DAG 执行 DAG 运行，因为 DAG 文件中的开始日期设置为昨天。这样，Airflow 就会赶上指定 DAG 的时间表。

示例 DAG 包含一项任务 print_dag_run_conf，该任务会在控制台中运行 echo 命令。此命令会输出有关 DAG 的元信息（DAG 运行的数字标识符）。

1. 在 DAG 标签页上，点击 composer_quickstart。系统会打开 DAG 的运行 标签页。

2. 在 DAG 运行列表中，点击第一个条目。

   

   图 2.composer_quickstart DAG 的 DAG 运行列表（点击可放大）

3. 系统会显示 DAG 运行详情，其中详细介绍了示例 DAG 的各项任务的相关信息。

   

   图 3.DAG 运行中执行的任务列表（点击可放大）

4. DAG 运行的日志 部分列出了 DAG 运行中所有任务的日志。您可以在日志中看到 echo 命令的输出。

   

   图 4.print_dag_run_conf 任务的日志 （点击可放大）

清理

为避免因本页中使用的资源导致您的 Google Cloud 账号产生费用，请按照以下步骤操作。

删除本教程中使用的资源 ：

1. 删除 Managed Airflow 环境：

   1. 在 Google Cloud 控制台中，前往环境页面。

      转到“环境”

   2. 选择 example-environment，并点击删除。

   3. 等待环境删除完成。

2. 删除环境的存储桶。删除 Managed Airflow 环境不会删除其存储桶。

   1. 在 Google Cloud 控制台中，前往 存储 > 浏览器 页面。

      转到“存储”>“浏览器”

   2. 选择环境的存储桶，然后点击删除。例如，此存储桶可以命名为 us-central1-example-environ-c1616fe8-bucket。

后续步骤

• 添加和更新 DAG
• 编写 DAG
• 替换 Airflow 配置
• 访问 Airflow 界面