Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
本教程介绍如何使用 Cloud Composer 创建 Apache Airflow DAG。DAG 会将 BigQuery 公共数据集中的数据与存储在 Cloud Storage 存储桶中的 CSV 文件相联接,然后运行 Dataproc Serverless 批量作业来处理联接的数据。
本教程中的 BigQuery 公共数据集是 ghcn_d,这是一个包含全球气候摘要的集成数据库。CSV 文件包含 1997 年至 2021 年期间美国节假日的日期和名称信息。
我们想通过 DAG 回答一个问题:“过去 25 年里,芝加哥感恩节的气温如何?”
目标
- 使用默认配置创建 Cloud Composer 环境
- 创建空 BigQuery 数据集
- 创建新的 Cloud Storage 存储桶
- 创建和运行包含以下任务的 DAG:
- 将外部数据集从 Cloud Storage 加载到 BigQuery
- 在 BigQuery 中联接两个数据集
- 运行数据分析 PySpark 作业
准备工作
启用 API
启用以下 API:
控制台
启用 Dataproc, Cloud Composer, BigQuery, Cloud Storage API。
gcloud
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs:
gcloud services enable dataproc.googleapis.comcomposer.googleapis.com bigquery.googleapis.com storage.googleapis.com
授予权限
向您的用户帐号授予以下角色和权限:
授予 BigQuery Data Owner (
roles/bigquery.dataOwner
) 角色以创建 BigQuery 数据集。授予 Storage Admin (
roles/storage.admin
) 角色以创建 Cloud Storage 存储桶。
创建和准备 Cloud Composer 环境
使用默认参数创建 Cloud Composer 环境:
- 选择一个美国境内的区域。
- 选择最新的 Cloud Composer 版本。
将以下角色授予 Cloud Composer 环境中使用的服务帐号,以便 Airflow 工作器成功运行 DAG 任务:
- BigQuery 用户 (
roles/bigquery.user
) - BigQuery 数据所有者 (
roles/bigquery.dataOwner
) - Service Account User (
roles/iam.serviceAccountUser
) - Dataproc Editor (
roles/dataproc.editor
) - Dataproc Worker (
roles/dataproc.worker
)
- BigQuery 用户 (
创建相关资源
使用以下参数创建一个空 BigQuery 数据集:
- 名称:
holiday_weather
- 区域:
US
- 名称:
在
US
多区域中创建新的 Cloud Storage 存储桶。运行以下命令,在要运行 Dataproc Serverless 的区域中的默认子网上启用专用 Google 访问通道,以满足网络要求。我们建议您使用与 Cloud Composer 环境相同的区域。
gcloud compute networks subnets update default \ --region DATAPROC_SERVERLESS_REGION \ --enable-private-ip-google-access
使用 Dataproc Serverless 处理数据处理
探索示例 PySpark 作业
以下代码是一个示例 PySpark 作业,它将温度从摄氏度的十分之一度转换为摄氏度。此作业会将数据集中的温度数据转换为其他格式。
将支持性文件上传到 Cloud Storage
如需上传 PySpark 文件和 holidays.csv
中存储的数据集,请执行以下操作:
将 data_analytics_process.py 保存到本地计算机。
将 holidays.csv 保存到本地机器。
在 Google Cloud 控制台中,前往 Cloud Storage 浏览器页面:
点击您之前创建的存储桶的名称。
在存储桶的对象标签页中,点击上传文件按钮,在随即显示的对话框中选择
data_analytics_process.py
和holidays.csv
,然后点击打开。
数据分析 DAG
探索示例 DAG
DAG 使用多个运算符来转换和统一数据:
GCSToBigQueryOperator
将 holidays.csv 文件从 Cloud Storage 提取到您之前创建的 BigQueryholidays_weather
数据集内的一个新表中。DataprocCreateBatchOperator
使用 Dataproc Serverless 创建和运行 PySpark 批量作业。BigQueryInsertJobOperator
将“日期”列的 holidays.csv 数据与 BigQuery 公共数据集 ghcn_d 中的天气数据相联接。BigQueryInsertJobOperator
任务是使用 for 循环动态生成的,这些任务位于TaskGroup
中,以便在 Airflow 界面的图表视图中提高可读性。
使用 Airflow 界面添加变量
在 Airflow 中,变量是一种通用方法,可用于以简单的键值对存储区的形式存储和检索任意设置或配置。此 DAG 使用 Airflow 变量存储常用值。如需将其添加到您的环境中,请执行以下操作:
转到管理 > 变量。
添加以下变量:
gcp_project
:您的项目 ID。gcs_bucket
:您之前创建的存储桶的名称(不带gs://
前缀)。gce_region
:您希望 Dataproc 作业满足 Dataproc 无服务器网络要求的区域。这是您之前启用了专用 Google 访问通道的区域。dataproc_service_account
:您的 Cloud Composer 环境的服务帐号。您可以在 Cloud Composer 环境的环境配置标签页上找到此服务帐号。
将该 DAG 上传到环境的存储桶
Cloud Composer 会安排位于您的环境存储桶的 /dags
文件夹中的 DAG。如需使用 Google Cloud 控制台上传 DAG,请执行以下操作:
在本地机器上,保存 data_analytics_dag.py。
在 Google Cloud 控制台中,前往环境页面。
在环境列表的 DAG 文件夹列中,点击 DAG 链接。系统会打开您的环境的 DAGs 文件夹。
点击上传文件。
在本地机器上选择
data_analytics_dag.py
,然后点击打开。
触发 DAG
在您的 Cloud Composer 环境中,点击 DAG 标签页。
点击进入 DAG ID
data_analytics_dag
。点击触发 DAG。
等待大约 5 到 10 分钟,直到出现表示任务已成功完成的绿色对勾标记。
验证 DAG 是否成功
在 Google Cloud 控制台中,转到 BigQuery 页面。
在浏览器面板中,点击您的项目名称。
点击
holidays_weather_joined
。点击“预览”可查看生成的表。请注意,值列中的数字为十分之一摄氏度。
点击
holidays_weather_normalized
。点击“预览”可查看生成的表。请注意,值列中的数字以摄氏度为单位。
深入了解 Dataproc Serverless(可选)
您可以使用更复杂的 PySpark 数据处理流程,试用此 DAG 的高级版本。请参阅 GitHub 上的适用于数据分析示例的 Dataproc 扩展程序。
清理
删除您为本教程创建的各个资源:
删除 Cloud Composer 环境,包括手动删除该环境的存储桶。