在 Cloud Composer 3 中运行 Apache Airflow DAG
Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3
本快速入门指南介绍了如何创建 Cloud Composer 环境 并在 Cloud Composer 3 中运行 Apache Airflow DAG。
如果您刚接触 Airflow,请参阅 Apache Airflow 中的 Airflow 概念教程 关于 Airflow 概念、对象和 使用情况。
如果您想改用 Google Cloud CLI,请参阅 在 Cloud Composer (Google Cloud CLI) 中运行 Apache Airflow DAG。
如果要使用 Terraform 创建环境,请参阅 创建环境 (Terraform)。
准备工作
- 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手,请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金,用于运行、测试和部署工作负载。
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
启用 Cloud Composer API。
-
如需获得完成本快速入门所需的权限,请让您的管理员为您授予项目的以下 IAM 角色:
-
如需查看创建和管理 Cloud Composer 环境,请执行以下操作:
-
环境和存储对象管理员 (
roles/composer.environmentAndStorageObjectAdmin
) -
Service Account User (
roles/iam.serviceAccountUser
)
-
环境和存储对象管理员 (
-
如需查看日志,请执行以下操作:
日志查看器 (
roles/logging.viewer
)
如需详细了解如何授予角色,请参阅管理访问权限。
-
如需查看创建和管理 Cloud Composer 环境,请执行以下操作:
创建环境
在 Google Cloud 控制台中,前往创建环境页面。
在名称字段中,输入
example-environment
。在位置下拉列表中,为 Cloud Composer 环境选择一个区域。本指南使用的是
us-central1
区域。对于其他环境配置选项,请使用提供的默认值。
点击创建,然后等待环境创建完成。
完成后,环境名称旁边会显示一个绿色对勾标记。
创建 DAG 文件
Airflow DAG 是有序任务的集合, 您想定期生成报表DAG 在标准 Python 文件中定义。
本指南使用的是 quickstart.py
文件中定义的示例 Airflow DAG。
此文件中的 Python 代码将执行以下操作:
- 创建一个 DAG
composer_sample_dag
。此 DAG 每天运行一次。 - 执行一项
print_dag_run_conf
任务。该任务使用 bash 运算符输出 DAG 运行的配置。
在本地机器上保存 quickstart.py
文件的副本:
将 DAG 文件上传到环境的存储桶
每个 Cloud Composer 环境都有一个 Cloud Storage
存储桶仅限 Cloud Composer 时间表中的 Airflow
位于此存储桶的 /dags
文件夹中的 DAG。
如需安排您的 DAG,请将 quickstart.py
从本地机器上传到您的
环境的 /dags
文件夹:
在 Google Cloud 控制台中,前往环境页面。
在环境列表中,点击您的环境名称
example-environment
。系统随即会打开环境详情页面。点击打开 DAGs 文件夹。系统随即会打开存储分区详情页面。
点击上传文件,然后选择
quickstart.py
副本。如需上传该文件,请点击打开。
查看 DAG
上传 DAG 文件后,Airflow 会执行以下操作:
- 解析您上传的 DAG 文件。系统可能需要几分钟时间 以供 Airflow 使用。
- 将 DAG 添加到可用 DAG 列表中。
- 根据您在 DAG 文件中提供的时间表执行 DAG。
检查您的 DAG 是否得到处理且没有任何错误,以及是否通过 在 DAG 界面中查看它。DAG 界面是 Cloud Composer 界面,可用于查看 Google Cloud 控制台中的 DAG 信息。Cloud Composer 还提供 访问 Airflow 界面,它是原生 Airflow Web 应用 界面。
等待大约 5 分钟,让 Airflow 有时间处理 DAG 文件 完成第一次 DAG 运行 (稍后介绍)。
在 Google Cloud 控制台中,前往环境页面。
在环境列表中,点击您的环境名称
example-environment
。系统随即会打开环境详情页面。转到 DAG 标签页。
检查是否存在
composer_quickstart
DAG 在 DAG 列表中。
查看 DAG 运行详情
DAG 的单次执行称为“DAG 运行”。Airflow 立即 针对示例 DAG 执行 DAG 运行,因为 DAG 文件中的开始日期是 设为昨天。这样,Airflow 能够与指定的 DAG 同步 时间表。
示例 DAG 包含一项 print_dag_run_conf
任务,该任务运行 echo
命令。此命令会输出有关 DAG 的元信息
(DAG 运行的数字标识符)。
在 DAG 标签页上,点击
composer_quickstart
。Runs 标签页 DAG 随即打开。在 DAG 运行列表中,点击第一个条目。
系统随即会显示 DAG 运行详情,其中详细说明了各个 示例 DAG 的任务。
DAG 运行日志部分列出了 DAG 运行中所有任务的日志。 您可以在日志中查看
echo
命令的输出。
清理
为避免因本页中使用的资源导致您的 Google Cloud 账号产生费用,请按照以下步骤操作。
删除本教程中使用的资源:
删除 Cloud Composer 环境:
在 Google Cloud 控制台中,前往环境页面。
选择
example-environment
,并点击删除。等待环境删除完成。
删除环境的存储桶。删除 Cloud Composer 环境不会删除其存储分区。
在 Google Cloud 控制台中,前往存储 >浏览器页面。
选择环境的存储桶,然后点击删除。例如,此存储桶可以命名为
us-central1-example-environ-c1616fe8-bucket
。
删除您的环境的 Redis 队列的永久性磁盘。删除 Cloud Composer 环境并不会删除其永久性磁盘。
在 Google Cloud 控制台中,前往 Compute Engine >磁盘。
选择环境的 Redis 队列永久性磁盘,然后点击删除。
例如,此磁盘可以命名为
gke-us-central1-exampl-pvc-b12055b6-c92c-43ff-9de9-10f2cc6fc0ee
。适用于 Cloud Composer 1 的磁盘始终为Standard persistent disk
类型,大小为 2 GB。
后续步骤