為因應客戶對 OSS 解決方案日益成長的需求，並進一步提升服務品質，Cloud Composer 將演變為 Managed Service for Apache Airflow。更名後，客戶將能更清楚瞭解我們的產品組合，同時也再次展現我們致力於打造最開放雲端生態系統的決心。

Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

使用 Azure 資料在 Google Cloud 中執行資料分析 DAG

Managed Airflow (第 3 代) | Managed Airflow (第 2 代) | Managed Airflow (舊版第 1 代)

本教學課程是在 Google Cloud中執行資料分析 DAG 的修改版本，說明如何將 Managed Airflow 環境連線至 Microsoft Azure，以使用儲存在該處的資料。說明如何使用 Managed Airflow 建立 Apache Airflow DAG。DAG 會合併 BigQuery 公開資料集和儲存在 Azure Blob 儲存體中的 CSV 檔案資料，然後執行 Managed Service for Apache Spark 批次工作，處理合併後的資料。

本教學課程使用的 BigQuery 公開資料集為 ghcn_d，這是全球氣候摘要的整合式資料庫。CSV 檔案包含 1997 年至 2021 年美國節慶的日期和名稱。

我們想透過 DAG 回答的問題是：「過去 25 年來，芝加哥的感恩節有多溫暖？」

目標

以預設設定建立 Managed Airflow 環境
在 Azure 中建立 Blob
建立空白的 BigQuery 資料集
建立新的 Cloud Storage bucket
建立並執行包含下列工作的 DAG：
- 將 Azure Blob 儲存體中的外部資料集載入 Cloud Storage
- 將外部資料集從 Cloud Storage 載入至 BigQuery
- 在 BigQuery 中聯結兩個資料集
- 執行資料分析 PySpark 工作

事前準備

啟用 API

啟用下列 API：

控制台

如果尚未啟用，請啟用 Managed Service for Apache Spark、Managed Airflow、BigQuery 和 Cloud Storage API。

啟用 API 時所需的角色

如要啟用 API，您必須具備 serviceusage.services.enable 權限。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「服務使用情形管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

啟用 API

gcloud

如果尚未啟用，請啟用 Managed Service for Apache Spark、Managed Airflow、BigQuery 和 Cloud Storage API：

啟用 API 時所需的角色

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

授予權限

將下列角色和權限授予使用者帳戶：

授予管理 Managed Airflow 環境和環境 bucket 的角色。
授予「BigQuery 資料擁有者」 (roles/bigquery.dataOwner) 角色，建立 BigQuery 資料集。
授予「儲存空間管理員」 (roles/storage.admin) 角色，建立 Cloud Storage bucket。

建立及準備 Managed Airflow 環境

使用預設參數建立 Managed Airflow 環境：
- 選擇美國境內的區域。
- 選擇最新Managed Airflow 版本。
注意： 本教學課程的 BigQuery 部分必須在 US 多重區域中執行。建議為 Managed Airflow 環境選擇美國區域，以降低費用和延遲時間，但如果 Managed Airflow 環境位於其他區域，仍可執行本教學課程。
將下列角色授予 Managed Airflow 環境中使用的服務帳戶，Airflow 工作者才能順利執行 DAG 工作：
- BigQuery 使用者 (roles/bigquery.user)
- BigQuery 資料擁有者 (roles/bigquery.dataOwner)
- 服務帳戶使用者 (roles/iam.serviceAccountUser)
- Dataproc 編輯者 (roles/dataproc.editor)
- Dataproc Worker (roles/dataproc.worker)

在 Google Cloud中建立及修改相關資源

在 Managed Airflow 環境中安裝 apache-airflow-providers-microsoft-azure PyPI 套件。
建立空白的 BigQuery 資料集，並使用下列參數：
- Name (名稱)：holiday_weather
- Region (區域)：US
在 US 多區域中建立新的 Cloud Storage bucket。
執行下列指令，在您要執行 Managed Service for Apache Spark 的區域中，啟用私人 Google 存取權，以符合網路需求。建議使用與 Managed Airflow 環境相同的區域。
```
gcloud compute networks subnets update default \
    --region DATAPROC_SERVERLESS_REGION \
    --enable-private-ip-google-access
```

在 Azure 中建立相關資源

建立儲存空間帳戶，並使用預設設定。
取得儲存空間帳戶的存取金鑰和連線字串。
在新建的儲存空間帳戶中，使用預設選項建立容器。
授予儲存體 Blob 委派人角色給上一個步驟中建立的容器。
將 holidays.csv 上傳至 Azure 入口網站，建立具有預設選項的區塊 Blob。
在 Azure 入口網站中，為您在上一個步驟建立的區塊 Blob 建立 SAS 權杖。
- 簽署方法：使用者委派金鑰
- 權限：讀取
- 允許的 IP 位址：無
- 允許的通訊協定：僅限 HTTPS

從 Managed Airflow 連線至 Azure

使用 Airflow UI 新增 Microsoft Azure連線：

依序前往「管理」>「連結」。
建立新連線並採用下列設定：
- 連線 ID： azure_blob_connection
- 連線類型： Azure Blob Storage
- Blob 儲存體登入：儲存體帳戶名稱
- Blob 儲存體金鑰：儲存體帳戶的存取金鑰
- Blob 儲存體帳戶連線字串：儲存體帳戶連線字串
- SAS 權杖：從 Blob 產生的 SAS 權杖

使用 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")

將 PySpark 檔案上傳至 Cloud Storage

如要將 PySpark 檔案上傳至 Cloud Storage，請按照下列步驟操作：

將 data_analytics_process.py 儲存到本機電腦。
前往 Google Cloud 控制台的「Cloud Storage 瀏覽器」頁面：

前往 Cloud Storage 瀏覽器
按一下您稍早建立的 bucket 名稱。
在 bucket 的「物件」分頁中，按一下「上傳檔案」按鈕，在隨即顯示的對話方塊中選取 data_analytics_process.py，然後按一下「開啟」。

資料分析 DAG

探索範例 DAG

DAG 會使用多個運算子轉換及合併資料：

AzureBlobStorageToGCSOperator 會將 Azure 區塊 Blob 中的 holidays.csv 檔案轉移至 Cloud Storage bucket。
GCSToBigQueryOperator 會將 Cloud Storage 中的 holidays.csv 檔案擷取至您先前建立的 BigQuery holidays_weather 資料集中的新資料表。
DataprocCreateBatchOperator 會使用 Managed Service for Apache Spark 建立及執行 PySpark 批次工作。
BigQueryInsertJobOperator 會根據「Date」欄，將 holidays.csv 中的資料與 BigQuery 公開資料集 ghcn_d 中的天氣資料合併。這些 BigQueryInsertJobOperator 工作是使用 for 迴圈動態產生，且位於 TaskGroup 中，可在 Airflow UI 的「Graph View」(圖表檢視畫面) 中提升可讀性。

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.providers.microsoft.azure.transfers.azure_blob_to_gcs import (
    AzureBlobStorageToGCSOperator,
)
from airflow.utils.task_group import TaskGroup

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

# 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}}"
PYSPARK_JAR = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
PROCESSING_PYTHON_FILE = f"gs://{BUCKET_NAME}/data_analytics_process.py"

# Azure configs
AZURE_BLOB_NAME = "{{var.value.azure_blob_name}}"
AZURE_CONTAINER_NAME = "{{var.value.azure_container_name}}"

BATCH_ID = "data-processing-{{ ts_nodash | lower}}"  # Dataproc serverless only allows lowercase characters
BATCH_CONFIG = {
    "pyspark_batch": {
        "jar_file_uris": [PYSPARK_JAR],
        "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(
    "azure_to_gcs_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    azure_blob_to_gcs = AzureBlobStorageToGCSOperator(
        task_id="azure_blob_to_gcs",
        # Azure args
        blob_name=AZURE_BLOB_NAME,
        container_name=AZURE_CONTAINER_NAME,
        wasb_conn_id="azure_blob_connection",
        filename=f"https://console.cloud.google.com/storage/browser/{BUCKET_NAME}/",
        # GCP args
        gcp_conn_id="google_cloud_default",
        object_name="holidays.csv",
        bucket_name=BUCKET_NAME,
        gzip=False,
        impersonation_chain=None,
    )

    create_batch = dataproc.DataprocCreateBatchOperator(
        task_id="create_batch",
        project_id=PROJECT_NAME,
        region=REGION,
        batch=BATCH_CONFIG,
        batch_id=BATCH_ID,
    )

    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):
            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",
            )

        azure_blob_to_gcs >> load_external_dataset >> bq_join_group >> create_batch

使用 Airflow UI 新增變數

在 Airflow 中，變數是儲存及擷取任意設定或設定的通用方式，可做為簡單的鍵值儲存區。這個 DAG 會使用 Airflow 變數儲存通用值。如要將這些項目新增至環境，請按照下列步驟操作：

從 Managed Airflow 控制台存取 Airflow 使用者介面。
依序前往「管理」> 變數。
新增下列變數：
- gcp_project：您的專案 ID。
- gcs_bucket：您先前建立的 bucket 名稱 (不含 gs:// 前置字串)。
- gce_region：您希望 Managed Service for Apache Spark 工作所在的區域，該區域必須符合 Managed Service for Apache Spark 網路需求。這是您先前啟用私人 Google 存取權的區域。
- dataproc_service_account：Managed Airflow 環境的服務帳戶。您可以在 Managed Airflow 環境的環境設定分頁中找到這個服務帳戶。
- azure_blob_name：您先前建立的 Blob 名稱。
- azure_container_name：您先前建立的容器名稱。

將 DAG 上傳至環境的 bucket

Managed Airflow 會排定環境值區中 /dags 資料夾內的 DAG。如要使用Google Cloud 控制台上傳 DAG，請按照下列步驟操作：

在本機電腦上儲存 azureblobstoretogcsoperator_tutorial.py。
前往 Google Cloud 控制台的「Environments」(環境) 頁面。

前往「環境」
在環境清單的「DAG 資料夾」欄中，按一下「DAG」連結。環境的 DAG 資料夾隨即開啟。
按一下「上傳檔案」。
選取本機電腦上的 azureblobstoretogcsoperator_tutorial.py，然後按一下「開啟」。

觸發 DAG

在 Managed Airflow 環境中，按一下「DAG」分頁標籤。
按一下 DAG ID azure_blob_to_gcs_dag。
按一下「觸發 DAG」。
等待約五到十分鐘，直到看到綠色勾號，表示工作已順利完成。

驗證 DAG 是否成功

前往 Google Cloud 控制台的「BigQuery」BigQuery頁面。

前往「BigQuery」
在「Explorer」面板中，按一下專案名稱。
按一下「holidays_weather_joined」。
按一下「預覽」即可查看產生的表格。請注意，「值」欄中的數字是以攝氏度數的十分之一為單位。
按一下「holidays_weather_normalized」。
按一下「預覽」即可查看產生的表格。請注意，「值」欄中的數字是以攝氏度為單位。

清除所用資源

刪除您為本教學課程建立的個別資源：

後續步驟

在 Google Cloud 中執行資料分析 DAG。
在 AWS 中執行資料分析 DAG。

使用 Azure 資料在 Google Cloud 中執行資料分析 DAG 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

目標

事前準備

啟用 API

控制台

gcloud

授予權限

建立及準備 Managed Airflow 環境

在 Google Cloud中建立及修改相關資源

在 Azure 中建立相關資源

從 Managed Airflow 連線至 Azure

使用 Managed Service for Apache Spark 處理資料

探索 PySpark 工作範例

將 PySpark 檔案上傳至 Cloud Storage

資料分析 DAG

探索範例 DAG

使用 Airflow UI 新增變數

將 DAG 上傳至環境的 bucket

觸發 DAG

驗證 DAG 是否成功

清除所用資源

後續步驟

使用 Azure 資料在 Google Cloud 中執行資料分析 DAG