2026 年 9 月 15 日，所有 Cloud Composer 1 和 Cloud Composer 2 2.0.x 版環境將達到預計的生命週期終止日，屆時您將無法使用這些環境。建議您規劃遷移至 Cloud Composer 3。

本頁面由 Cloud Translation API 翻譯而成。

使用 AWS 的資料，在 Google Cloud 中執行資料分析 DAG

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本教學課程是「在 Cloud Composer 中執行資料分析 DAG Google Cloud」的修改版本，說明如何將 Cloud Composer 環境連線至 Amazon Web Services，以使用儲存在該處的資料。說明如何使用 Cloud Composer 建立 Apache Airflow DAG。DAG 會合併 BigQuery 公開資料集和儲存在 Amazon Web Services (AWS) S3 值區中的 CSV 檔案資料，然後執行 Dataproc Serverless 批次工作，處理合併後的資料。

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

我們想透過 DAG 回答的問題是：「過去 25 年來，芝加哥在感恩節當天的氣溫如何？」

目標

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

事前準備

在 AWS 中管理權限

建立 AWS 帳戶。
請按照「Creating policies with the visual editor section」(使用視覺化編輯器建立政策) 一節的 Creating IAM Policies AWS tutorial (建立 IAM 政策 AWS 教學課程) 操作，為 AWS S3 建立自訂 IAM 政策，並採用下列設定：
- 服務：S3
- ListAllMyBuckets (s3:ListAllMyBuckets)，用於查看 S3 儲存空間
- CreateBucket (s3:CreateBucket)：用於建立 bucket
- PutBucketOwnershipControls (s3:PutBucketOwnershipControls)，用於建立 bucket
- ListBucket (s3:ListBucket)：授予列出 S3 bucket 中物件的權限
- PutObject (s3:PutObject)：將檔案上傳至 bucket
- GetBucketVersioning (s3:GetBucketVersioning)，用於刪除 bucket 中的物件
- DeleteObject (s3:DeleteObject)：用於刪除值區中的物件
- ListBucketVersions (s3:ListBucketVersions)，用於刪除值區
- DeleteBucket (s3:DeleteBucket)：用於刪除值區
- 資源：在「bucket」和「object」旁邊選擇「Any」，即可授予該類型任何資源的權限。
- 標記：無
- 名稱：TutorialPolicy
如要進一步瞭解上述各項設定，請參閱 Amazon S3 支援的動作清單。
將 TutorialPolicy IAM 政策新增至身分

啟用 API

啟用下列 API：

主控台

Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs.

Enable the APIs

gcloud

Enable the Dataproc, Cloud Composer, BigQuery, Cloud Storage APIs:

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

授予權限

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

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

建立及準備 Cloud Composer 環境

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

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

在 Cloud Composer 環境中安裝 apache-airflow-providers-amazon PyPI 套件。
建立空白的 BigQuery 資料集，並使用下列參數：
- Name (名稱)：holiday_weather
- Region (區域)：US
在 US 多區域中建立新的 Cloud Storage 值區。
執行下列指令，在要執行 Dataproc Serverless 的區域中，對預設子網路啟用私人 Google 存取權，以滿足網路需求。建議使用與 Cloud Composer 環境相同的區域。
```
gcloud compute networks subnets update default \
    --region DATAPROC_SERVERLESS_REGION \
    --enable-private-ip-google-access
```

在 AWS 中建立相關資源

在偏好的區域中，使用預設設定建立 S3 bucket。

從 Cloud Composer 連線至 AWS

取得 AWS 存取金鑰 ID 和存取密鑰
使用 Airflow UI 新增 AWS S3連線：
1. 依序前往「管理」>「連結」。
2. 建立新連線並加入以下設定：
  - 連線 ID： aws_s3_connection
  - 連線類型： Amazon S3
  - 額外內容： {"aws_access_key_id":"your_aws_access_key_id", "aws_secret_access_key": "your_aws_secret_access_key"}

使用 Dataproc Serverless 處理資料

探索 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 名稱。
在值區的「物件」分頁中，按一下「上傳檔案」按鈕，在出現的對話方塊中選取 data_analytics_process.py，然後按一下「開啟」。

將 CSV 檔案上傳至 AWS S3

如要上傳 holidays.csv 檔案，請按照下列步驟操作：

將 holidays.csv 儲存在本機電腦上。
按照 AWS 指南的說明，將檔案上傳至值區。

資料分析 DAG

探索範例 DAG

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

S3ToGCSOperator 將 AWS S3 bucket 中的 holidays.csv 檔案轉移至 Cloud Storage bucket。
GCSToBigQueryOperator 會從 Cloud Storage 將 holidays.csv 檔案擷取到您先前建立的 BigQuery holidays_weather 資料集中的新資料表。
DataprocCreateBatchOperator 會使用 Dataproc Serverless 建立及執行 PySpark 批次工作。
BigQueryInsertJobOperator 會將 holidays.csv 中的資料，與 BigQuery 公開資料集 ghcn_d 中的氣象資料，依據「Date」欄位進行聯結。BigQueryInsertJobOperator 工作是使用 for 迴圈動態產生，這些工作位於 TaskGroup 中，方便在 Airflow UI 的圖表檢視畫面中閱讀。

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.google.cloud.transfers.s3_to_gcs import S3ToGCSOperator
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"

# S3 configs
S3_BUCKET_NAME = "{{var.value.s3_bucket}}"

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(
    "s3_to_gcs_dag",
    # Continue to run DAG once per day
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args,
) as dag:
    s3_to_gcs_op = S3ToGCSOperator(
        task_id="s3_to_gcs",
        bucket=S3_BUCKET_NAME,
        gcp_conn_id="google_cloud_default",
        aws_conn_id="aws_s3_connection",
        dest_gcs=f"gs://{BUCKET_NAME}",
    )

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

        s3_to_gcs_op >> load_external_dataset >> bq_join_group >> create_batch

使用 Airflow UI 新增變數

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

從 Cloud Composer 控制台存取 Airflow UI。
依序前往「管理」>「變數」。
新增下列變數：
- s3_bucket：您先前建立的 S3 值區名稱。
- gcp_project：您的專案 ID。
- gcs_bucket：您先前建立的值區名稱 (不含 gs:// 前置字串)。
- ：您要執行 Dataproc 作業的區域，該作業必須符合 Dataproc Serverless 網路需求。gce_region這是您先前啟用私人 Google 存取權的區域。
- dataproc_service_account：Cloud Composer 環境的服務帳戶。您可以在 Cloud Composer 環境的環境設定分頁中找到這個服務帳戶。

將 DAG 上傳至環境的值區

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

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

前往「環境」
在環境清單的「DAG folder」(DAG 資料夾) 欄中，按一下「DAGs」(DAG) 連結。系統會開啟環境的 DAG 資料夾。
按一下「上傳檔案」。
選取本機電腦上的 s3togcsoperator_tutorial.py，然後按一下「開啟」。

觸發 DAG

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

驗證 DAG 是否成功

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

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

清除所用資源

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