在 Google Cloud 中运行数据分析 DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本教程介绍如何使用 Cloud Composer 创建 Apache Airflow DAG。该 DAG 会联接来自 BigQuery 公共数据集和存储在 Cloud Storage 存储桶中的 CSV 文件的数据，然后运行 Managed Service for Apache Spark 批量作业来处理联接的数据。

本教程中的 BigQuery 公共数据集是 ghcn_d，这是一个全球气候汇总集成数据库。CSV 文件包含 1997 年至 2021 年美国节假日的日期和名称信息 。

我们希望使用 DAG 回答的问题是：“过去 25 年来，芝加哥感恩节有多温暖？”

目标

• 在默认配置中创建一个 Cloud Composer 环境
• 创建一个空的 BigQuery 数据集
• 创建一个新的 Cloud Storage 存储桶
• 创建并运行一个包含以下任务的 DAG：
  • 将外部数据集从 Cloud Storage 加载到 BigQuery
  • 在 BigQuery 中联接两个数据集
  • 运行数据分析 PySpark 作业

准备工作

启用 API

启用以下 API：

控制台

启用 Managed Service for Apache Spark、Cloud Composer、BigQuery、Cloud Storage API。

启用 API 所需的角色

如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予 角色。

启用 API

gcloud

启用 Managed Service for Apache Spark、Cloud Composer、BigQuery、Cloud Storage API：

启用 API 所需的角色

如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予 角色。

gcloud services enable dataproc.googleapis.com  composer.googleapis.com  bigquery.googleapis.com  storage.googleapis.com

授予权限

向您的用户账号授予以下角色和权限：

• 授予用于管理 Cloud Composer 环境和环境存储分区的角色。

• 授予 BigQuery Data Owner (roles/bigquery.dataOwner) 角色以 创建 BigQuery 数据集。

• 授予 Storage Admin (roles/storage.admin) 角色以 创建 Cloud Storage 存储桶。

创建并准备 Cloud Composer 环境

1. 使用默认参数创建 Cloud Composer 环境 ：

   • 选择美国境内的区域。
   • 选择最新 Cloud Composer 版本。
   注意 ：本教程的 BigQuery 部分必须在 US 多区域中运行。我们建议您为 Cloud Composer 环境选择美国境内的区域，以降低费用和延迟时间，但如果您的 Cloud Composer 环境位于其他区域，本教程仍然可以运行。

2. 向 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)

创建相关资源

1. 创建一个空的 BigQuery 数据集 使用以下参数：

   • 名称: holiday_weather
   • 区域：US

2. 创建一个新的 Cloud Storage 存储桶 在 US 多区域中。

3. 运行以下命令，在您想要运行 Managed Service for Apache Spark 的区域中的默认子网上 启用专用 Google 访问通道 ，以满足 网络要求。我们建议您使用与 Cloud Composer 环境相同的区域。

   gcloud compute networks subnets update default \
       --region DATAPROC_SERVERLESS_REGION \
       --enable-private-ip-google-access
   

注意 ：空的 BigQuery 数据集必须位于美国区域，以满足 转移 BigQuery 公共数据集 ghcn_d 的 同地协作要求。

使用 Managed Service for Apache Spark 进行数据处理

探索示例 PySpark 作业

以下代码是一个示例 PySpark 作业，用于将温度从摄氏度十分之一转换为摄氏度。此作业会将数据集中的温度数据转换为不同的格式。

import sys


from py4j.protocol import Py4JJavaError
from pyspark.sql import SparkSession
from pyspark.sql.functions import col


if __name__ == "__main__":
    BUCKET_NAME = sys.argv[1]
    READ_TABLE = sys.argv[2]
    WRITE_TABLE = sys.argv[3]

    # Create a SparkSession, viewable via the Spark UI
    spark = SparkSession.builder.appName("data_processing").getOrCreate()

    # Load data into dataframe if READ_TABLE exists
    try:
        df = spark.read.format("bigquery").load(READ_TABLE)
    except Py4JJavaError as e:
        raise Exception(f"Error reading {READ_TABLE}") from e

    # Convert temperature from tenths of a degree in celsius to degrees celsius
    df = df.withColumn("value", col("value") / 10)
    # Display sample of rows
    df.show(n=20)

    # Write results to GCS
    if "--dry-run" in sys.argv:
        print("Data will not be uploaded to BigQuery")
    else:
        # Set GCS temp location
        temp_path = BUCKET_NAME

        # Saving the data to BigQuery using the "indirect path" method and the spark-bigquery connector
        # Uses the "overwrite" SaveMode to ensure DAG doesn't fail when being re-run
        # See https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#save-modes
        # for other save mode options
        df.write.format("bigquery").option("temporaryGcsBucket", temp_path).mode(
            "overwrite"
        ).save(WRITE_TABLE)
        print("Data written to BigQuery")

将支持文件上传到 Cloud Storage

如需上传 PySpark 文件和存储在 holidays.csv 中的数据集，请执行以下操作：

1. 将 data_analytics_process.py 保存到本地机器。

2. 将 holidays.csv 保存到本地机器。

3. 在 Google Cloud 控制台中，前往 Cloud Storage 浏览器 页面：

   前往 Cloud Storage 浏览器

4. 点击您之前创建的存储桶的名称。

5. 在存储桶的对象 标签页中，点击上传文件 按钮，在随即显示的对话框中选择 data_analytics_process.py 和 holidays.csv，然后点击打开 。

数据分析 DAG

探索示例 DAG

该 DAG 使用多个运算符来转换和统一数据：

• GCSToBigQueryOperator 会将 holidays.csv 文件从 Cloud Storage 提取到您之前创建的 BigQuery holidays_weather 数据集中的新表。

• The DataprocCreateBatchOperator 使用 Managed Service for Apache Spark 创建并运行 PySpark 批量作业。

• BigQueryInsertJobOperator 会将 holidays.csv 中“Date”列的数据与 BigQuery 公共数据集 ghcn_d 中的天气数据联接。BigQueryInsertJobOperator 任务是 使用 for 循环动态生成的，这些任务位于 TaskGroup 中，以便在 Airflow 界面的 Graph View 中获得更好的可读性。

重要提示 ：DataprocCreateBatchOperator 在 apache-airflow-providers-google 软件包版本 6.2.0 及更高版本中提供。从 Cloud Composer 版本 1.17.10 和 2.0.3 开始，您的环境中 预安装了此软件包的必需版本。

import datetime

from airflow import models
from airflow.providers.google.cloud.operators import dataproc
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import (
    GCSToBigQueryOperator,
)
from airflow.utils.task_group import TaskGroup

PROJECT_NAME = "{{var.value.gcp_project}}"

# BigQuery configs
BQ_DESTINATION_DATASET_NAME = "holiday_weather"
BQ_DESTINATION_TABLE_NAME = "holidays_weather_joined"
BQ_NORMALIZED_TABLE_NAME = "holidays_weather_normalized"

# Dataproc configs
BUCKET_NAME = "{{var.value.gcs_bucket}}"
PROCESSING_PYTHON_FILE = f"gs://{BUCKET_NAME}/data_analytics_process.py"

BATCH_ID = "data-processing-{{ ts_nodash | lower}}"  # Dataproc serverless only allows lowercase characters
BATCH_CONFIG = {
    "runtime_config": {"version": "1.1"},
    "pyspark_batch": {
        "main_python_file_uri": PROCESSING_PYTHON_FILE,
        "args": [
            BUCKET_NAME,
            f"{BQ_DESTINATION_DATASET_NAME}.{BQ_DESTINATION_TABLE_NAME}",
            f"{BQ_DESTINATION_DATASET_NAME}.{BQ_NORMALIZED_TABLE_NAME}",
        ],
    },
    "environment_config": {
        "execution_config": {
            "service_account": "{{var.value.dataproc_service_account}}"
        }
    },
}

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()
)

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    "start_date": yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    "email_on_failure": False,
    "email_on_retry": False,
}

with models.DAG(
    "data_analytics_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    create_batch = dataproc.DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_NAME,
        region="{{ var.value.gce_region }}",
        batch=BATCH_CONFIG,
        batch_id=BATCH_ID,
    )
    # This data is static and it is safe to use WRITE_TRUNCATE
    # to reduce chance of 409 duplicate errors
    load_external_dataset = GCSToBigQueryOperator(
        task_id="run_bq_external_ingestion",
        bucket=BUCKET_NAME,
        source_objects=["holidays.csv"],
        destination_project_dataset_table=f"{BQ_DESTINATION_DATASET_NAME}.holidays",
        source_format="CSV",
        schema_fields=[
            {"name": "Date", "type": "DATE"},
            {"name": "Holiday", "type": "STRING"},
        ],
        skip_leading_rows=1,
        write_disposition="WRITE_TRUNCATE",
    )

    with TaskGroup("join_bq_datasets") as bq_join_group:
        for year in range(1997, 2022):
            # BigQuery configs
            BQ_DATASET_NAME = f"bigquery-public-data.ghcn_d.ghcnd_{str(year)}"
            BQ_DESTINATION_TABLE_NAME = "holidays_weather_joined"
            # Specifically query a Chicago weather station
            WEATHER_HOLIDAYS_JOIN_QUERY = f"""
            SELECT Holidays.Date, Holiday, id, element, value
            FROM `{PROJECT_NAME}.holiday_weather.holidays` AS Holidays
            JOIN (SELECT id, date, element, value FROM {BQ_DATASET_NAME} AS Table WHERE Table.element="TMAX" AND Table.id="USW00094846") AS Weather
            ON Holidays.Date = Weather.Date;
            """

            # for demo purposes we are using WRITE_APPEND
            # but if you run the DAG repeatedly it will continue to append
            # Your use case may be different, see the Job docs
            # https://cloud.google.com/bigquery/docs/reference/rest/v2/Job
            # for alternative values for the writeDisposition
            # or consider using partitioned tables
            # https://cloud.google.com/bigquery/docs/partitioned-tables
            bq_join_holidays_weather_data = BigQueryInsertJobOperator(
                task_id=f"bq_join_holidays_weather_data_{str(year)}",
                configuration={
                    "query": {
                        "query": WEATHER_HOLIDAYS_JOIN_QUERY,
                        "useLegacySql": False,
                        "destinationTable": {
                            "projectId": PROJECT_NAME,
                            "datasetId": BQ_DESTINATION_DATASET_NAME,
                            "tableId": BQ_DESTINATION_TABLE_NAME,
                        },
                        "writeDisposition": "WRITE_APPEND",
                    }
                },
                location="US",
            )

        load_external_dataset >> bq_join_group >> create_batch

使用 Airflow 界面添加变量

在 Airflow 中， 变量 是一种通用方式，用于将任意设置或 配置存储为简单的键值存储并从中检索。此 DAG 使用 Airflow 变量来存储常见值。如需将它们添加到您的环境中，请执行以下操作：

1. 从 Cloud Composer 控制台访问 Airflow 界面。

2. 前往管理员 > 变量。

3. 添加以下变量：

   • gcp_project：您的项目 ID。

   • gcs_bucket：您之前创建的存储桶的名称（不带 gs:// 前缀）。

   • gce_region：您希望您的 Managed Service for Apache Spark 作业满足 Managed Service for Apache Spark 网络要求的区域。 这是您之前启用专用 Google 访问通道的区域。

   • dataproc_service_account：Cloud Composer 环境的服务帐号。您可以在 Cloud Composer 环境的环境配置标签页中找到此服务账号。

将 DAG 上传到环境的存储桶

Cloud Composer 会安排位于环境的存储桶中的 /dags 文件夹中的 DAG。如需使用 Google Cloud 控制台上传 DAG，请执行以下操作：

1. 在本地机器上，保存 data_analytics_dag.py。

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

   前往“环境”

3. 在环境列表中，点击 DAG 文件夹 列中的 DAG 链接。系统会打开您环境的 DAG 文件夹。

4. 点击上传文件 。

5. 在本地机器上选择 data_analytics_dag.py，然后点击打开 。

触发 DAG

1. 在 Cloud Composer 环境中，点击 DAG 标签页。

2. 点击 DAG ID data_analytics_dag。

3. 点击触发 DAG 。

4. 等待大约 5 到 10 分钟，直到看到绿色对勾，表示任务已成功完成。

验证 DAG 是否成功

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

   前往 BigQuery

2. 在浏览器 面板中，点击您的项目名称。

3. 点击 holidays_weather_joined。

4. 点击“预览”以查看结果表。请注意，值列中的数字以摄氏度十分之一为单位。

5. 点击 holidays_weather_normalized。

6. 点击“预览”以查看结果表。请注意，值列中的数字以摄氏度为单位。

使用 Managed Service for Apache Spark 深入了解（可选）

您可以尝试使用更复杂的 PySpark 数据处理流程的此 DAG 的高级版本。请参阅 GitHub 上的 Data Analytics Example 的 Managed Service for Apache Spark 扩展程序 。

清理

删除您为本教程创建的各个资源：

• 删除 Cloud Storage 存储桶，该存储分区是您 为本教程创建的。

• 删除 BigQuery 数据集。

• 删除 Cloud Composer 环境，包括 手动删除环境的存储桶。

后续步骤

• 在 Google Cloud 中使用来自 AWS 的数据运行数据分析 DAG。
• 在 Azure 中运行数据分析 DAG。