Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1
本教程是对在 Google Cloud 中运行数据分析 DAG Google Cloud 的修改,介绍了如何将 Cloud Composer 环境连接到 Microsoft Azure,以便利用其中存储的数据。该教程介绍了如何使用 Cloud Composer 创建 Apache Airflow DAG。DAG 会联接 BigQuery 公共数据集和存储在 Azure Blob Storage 中的 CSV 文件中的数据,然后运行 Dataproc Serverless 批处理作业来处理联接后的数据。
本教程中的 BigQuery 公共数据集是 ghcn_d,它是一个集成的全球气候摘要数据库。CSV 文件包含 1997 年至 2021 年美国节日的日期和名称的相关信息。
我们希望使用 DAG 回答的问题是:“过去 25 年来,芝加哥在感恩节当天天气有多暖?”
目标
- 在默认配置中创建 Cloud Composer 环境
- 在 Azure 中创建 Blob
- 创建一个空的 BigQuery 数据集
- 创建新的 Cloud Storage 存储分区
- 创建和运行包含以下任务的 DAG:
- 将外部数据集从 Azure Blob Storage 加载到 Cloud Storage
- 将外部数据集从 Cloud Storage 加载到 BigQuery
- 在 BigQuery 中联接两个数据集
- 运行数据分析 PySpark 作业
准备工作
启用 API
启用以下 API:
控制台
Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs.
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 User (
roles/bigquery.user
) - BigQuery Data Owner (
roles/bigquery.dataOwner
) - Service Account User (
roles/iam.serviceAccountUser
) - Dataproc Editor (
roles/dataproc.editor
) - Dataproc Worker (
roles/dataproc.worker
)
- BigQuery User (
在 Google Cloud中创建和修改相关资源
在 Cloud Composer 环境中安装
apache-airflow-providers-microsoft-azure
PyPI 软件包。使用以下参数创建一个空的 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
在 Azure 中创建相关资源
使用默认设置创建存储账号。
在您新创建的存储空间账号中使用默认选项创建容器。
为在上一步中创建的容器授予 Storage Blob Delegator 角色。
上传 holidays.csv,以便在 Azure 门户中使用默认选项创建一个分块 blob。
在 Azure 门户中,为您在上一步中创建的块 blob 创建 SAS 令牌。
- 签名方法:用户委托密钥
- 权限:读取
- 允许使用的 IP 地址:无
- 允许的协议:仅限 HTTPS
从 Cloud Composer 连接到 Azure
使用 Airflow 界面添加 Microsoft Azure 连接:
依次选择管理 > 关联。
使用以下配置创建新连接:
- 连接 ID:
azure_blob_connection
- 连接类型:
Azure Blob Storage
- Blob Storage 登录信息:您的存储账号名称
- Blob Storage 密钥:存储账号的访问密钥
- Blob Storage 账号连接字符串:您的存储账号连接字符串
- SAS 令牌:从您的 blob 生成的 SAS 令牌
- 连接 ID:
使用 Dataproc Serverless 进行数据处理
探索 PySpark 作业示例
以下代码是一个 PySpark 作业示例,用于将温度从十分之一摄氏度转换为摄氏度。此作业会将数据集中的温度数据转换为其他格式。
将 PySpark 文件上传到 Cloud Storage
如需将 PySpark 文件上传到 Cloud Storage,请执行以下操作:
将 data_analytics_process.py 保存到本地机器。
在 Google Cloud 控制台中,前往 Cloud Storage 浏览器页面:
点击您之前创建的存储分区的名称。
在相应存储分区的对象标签页中,点击上传文件按钮,在随即显示的对话框中选择
data_analytics_process.py
,然后点击打开。
数据分析 DAG
探索示例 DAG
DAG 使用多个运算符来转换和统一数据:
AzureBlobStorageToGCSOperator
会将 holidays.csv 文件从您的 Azure 块 blob 传输到您的 Cloud Storage 存储分区。GCSToBigQueryOperator
会将 Cloud Storage 中的 holidays.csv 文件提取到您之前创建的 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 Serverless 网络要求的 Dataproc 作业的目标区域。这是您之前启用专用 Google 访问通道的区域。dataproc_service_account
:Cloud Composer 环境的服务账号。您可以在 Cloud Composer 环境的“环境配置”标签页中找到此服务账号。azure_blob_name
:您之前创建的 blob 的名称。azure_container_name
:您之前创建的容器的名称。
将 DAG 上传到您环境的存储分区
Cloud Composer 会安排位于环境存储分区的 /dags
文件夹中的 DAG。如需使用 Google Cloud 控制台上传 DAG,请执行以下操作:
在本地机器上,保存 azureblobstoretogcsoperator_tutorial.py。
在 Google Cloud 控制台中,前往环境页面。
在环境列表的 DAG 文件夹列中,点击 DAG 链接。系统会打开您环境的 DAG 文件夹。
点击上传文件。
在本地机器上选择
azureblobstoretogcsoperator_tutorial.py
,然后点击打开。
触发 DAG
在 Cloud Composer 环境中,点击 DAG 标签页。
点击 DAG ID
azure_blob_to_gcs_dag
。点击触发 DAG。
等待大约 5 到 10 分钟,直到您看到绿色对勾,表示任务已成功完成。
验证 DAG 是否成功
在 Google Cloud 控制台中,转到 BigQuery 页面。
在浏览器面板中,点击您的项目名称。
点击
holidays_weather_joined
。点击“预览”以查看生成的表格。请注意,值列中的数字以摄氏度为单位,小数点后有一位。
点击
holidays_weather_normalized
。点击“预览”以查看生成的表格。请注意,值列中的数字以摄氏度为单位。
清理
删除您为本教程创建的各个资源:
删除 Cloud Composer 环境,包括手动删除环境的存储分区。