Cloud Composer 1 はメンテナンス後モードです。Google は、Airflow の新しいバージョン、バグの修正、セキュリティアップデートなど、Cloud Composer 1 への今後のアップデートはリリースしません。 Cloud Composer 2 への移行の計画をおすすめします。

AWS のデータを使用して Google Cloud でデータ分析 DAG を実行する

Cloud Composer 1 | Cloud Composer 2

このチュートリアルは、Cloud Composer 環境を Amazon Web Services に接続して、そこに格納されているデータを利用する方法を説明している、Google Cloud でデータ分析 DAG を実行するの修正版です。Cloud Composer を使用して Apache Airflow DAG を作成する方法について説明します。DAG は、BigQuery 一般公開データセットと Amazon Web Services（AWS）S3 バケットに保存されている CSV ファイルのデータを結合してから、Dataproc サーバーレスバッチジョブを実行して結合されたデータを処理します。

このチュートリアルの BigQuery 一般公開データセットは、世界中の気候統合データベースである ghcn_d です。CSV ファイルには、1997 年から 2021 年までの米国の休日の日付と名前に関する情報が含まれています。

DAG を使用して答えを得たい質問は、「この 25 年間で感謝祭のシカゴはどのくらい温かかったか」というものです。

目標

デフォルト構成で Cloud Composer 環境を作成する
AWS S3 でバケットを作成する
空の BigQuery データセットを作成する
Cloud Storage バケットを新規作成する
次のタスクを含む DAG を作成、実行します。
- S3 から Cloud Storage に外部データセットを読み込む
- Cloud Storage から BigQuery に外部データセットを読み込む
- BigQuery で 2 つのデータセットを結合する
- データ分析の PySpark ジョブを実行する

準備

AWS で権限を管理する

AWS アカウントを作成する。
IAM ポリシー作成の AWS チュートリアルの「ビジュアルエディタでのポリシーの作成」セクションに従って、次の構成で AWS S3 用にカスタマイズされた IAM ポリシーを作成します。
- サービス: S3
- ListAllMyBuckets（s3:ListAllMyBuckets）、S3 バケット表示のため
- CreateBucket（s3:CreateBucket）、バケット作成のため
- PutBucketPersistenceControls（s3:PutBucketOwnershipControls）、バケット作成のため
- ListBucket（s3:ListBucket）、S3 バケット内のオブジェクトを一覧表示する権限を付与するため
- PutObject（s3:PutObject）: バケットにファイルをアップロードするため
- GetBucketVersioning（s3:GetBucketVersioning）、バケット内のオブジェクトを削除するため
- DeleteObject（s3:DeleteObject）: バケット内のオブジェクトを削除するため
- ListBucketVersions（s3:ListBucketVersions）、バケット削除のため
- DeleteBucket（s3:DeleteBucket）、バケット削除のため
- リソース: [bucket] と [object] の横にある [Any] を選択して、そのタイプのすべてのリソースに権限を付与します。
- タグ: なし
- 名前: TutorialPolicy
上記の各構成の詳細については、Amazon S3 でサポートされるアクションのリストをご覧ください。
自分の ID に 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 環境と環境バケットを管理するためのロールを付与します。
BigQuery データオーナー（roles/bigquery.dataOwner）のロールを付与して、BigQuery データセットを作成します。
ストレージ管理者（roles/storage.admin）のロールを付与して、Cloud Storage バケットを作成します。

Cloud Composer 環境を作成して準備する

デフォルトのパラメータで Cloud Composer 環境を作成します。
- 米国に置かれたリージョンを選択します。
- 最新の Cloud Composer バージョンを選択します。
注: このチュートリアルの BigQuery 部分は US マルチリージョンで実行する必要があります。費用とレイテンシを軽減するために、Cloud Composer 環境用に US リージョンを選択することをおすすめしますが、Cloud Composer 環境が別のリージョンにある場合でもチュートリアルを実行できます。
Airflow ワーカーが DAG タスクを正常に実行するために、Cloud Composer 環境で使用されるサービスアカウントに次のロールを付与します。
- BigQuery ユーザー（roles/bigquery.user）
- BigQuery データオーナー（roles/bigquery.dataOwner）
- サービスアカウントユーザー（roles/iam.serviceAccountUser）
- Dataproc 編集者（roles/dataproc.editor）
- Dataproc ワーカー（roles/dataproc.worker）

Google Cloud で関連リソースを作成して変更する

Cloud Composer 環境に apache-airflow-providers-amazon PyPI パッケージをインストールします。
次のパラメータを使用して空の BigQuery データセットを作成します。
- 名前: holiday_weather
- リージョン: US
US マルチリージョンで新しい Cloud Storage バケットを作成します。
次のコマンドを実行して、ネットワーキングの要件を満たすために Dataproc サーバーレスを実行するリージョン内のデフォルトサブネットで限定公開の Google アクセスを有効にします。Cloud Composer 環境と同じリージョンを使用することをおすすめします。
```
gcloud compute networks subnets update default \
    --region DATAPROC_SERVERLESS_REGION \
    --enable-private-ip-google-access
```

AWS で関連リソースを作成する

任意のリージョンにデフォルト設定で S3 バケットを作成します。

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 サーバーレスを使用したデータ処理

PySpark ジョブの例を確認する

次のコードは、温度を摂氏 10 分の 1 の度数から摂氏度数に変換する 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 にアップロードします。

data_analytics_process.py をローカルマシンに保存します。
Google Cloud コンソールで、Cloud Storage ブラウザページに移動します。

Cloud Storage ブラウザに移動
前の手順で作成したバケットの名前をクリックします。
バケットの [オブジェクト] タブで、[ファイルをアップロード] ボタンをクリックし、表示されたダイアログで data_analytics_process.py を選択し、[開く] をクリックします。

CSV ファイルを AWS S3 にアップロードする

holidays.csv ファイルをアップロードするには:

ローカルマシンに holidays.csv を保存する。
AWS ガイドに従って、ファイルをバケットにアップロードする。

データ分析 DAG

DAG の例を確認する

DAG では複数の演算子を使用してデータを変換し、統合します。

S3ToGCSOperator は、holidays.csv ファイルを AWS S3 バケットから Cloud Storage バケットに転送します。
GCSToBigQueryOperatorは、Cloud Storage からの holiday.csv ファイルを、前の手順で作成した BigQuery holidays_weather データセット内の新しいテーブルに取り込みます。
DataprocCreateBatchOperator は、Dataproc サーバーレスを使用して PySpark バッチジョブを作成し、実行します。
BigQueryInsertJobOperator は、[Date] 列の holidays.csv のデータを BigQuery 一般公開データセット ghcn_d の気象データと結合します。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 における変数は、任意の設定や構成をシンプルな Key-Value ストアとして保存および取得するためのユニバーサルな方法です。この DAG では Airflow 変数を使用して共通の値を保存します。ご利用の環境に追加するには:

Cloud Composer コンソールから Airflow UI にアクセスします。
[管理] > [変数] に移動します。
次の変数を追加します。
- s3_bucket: 前の手順で作成した S3 バケットの名前。
- 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 をアップロードするには:

ローカルマシンで、s3togcsoperator_tutorial.py を保存します。
Google Cloud Console で [環境] ページに移動します。

[環境] に移動
環境のリストにある [DAG フォルダ] 列で、[DAG] リンクをクリックします。環境の DAG フォルダが開きます。
[ファイルをアップロード] をクリックします。
ローカルマシン上の s3togcsoperator_tutorial.py を選択して、[開く] をクリックします。

DAG をトリガーする

Cloud Composer 環境で [DAG] タブをクリックします。
DAG ID s3_to_gcs_dag をクリックします。
[DAG をトリガー] をクリックします。
タスクが正常に完了したことを示す緑色のチェックマークが表示されるまで、5～10 分待ちます。

DAG の成功を検証する

Google Cloud コンソールで [BigQuery] ページに移動します。

BigQuery に移動
[エクスプローラ] パネルでプロジェクト名をクリックします。
[holidays_weather_joined] をクリックします。
[プレビュー] をクリックすると、結果の表が表示されます。値列の数値は、10 分の 1 の摂氏度数です。
[holidays_weather_normalized] をクリックします。
[プレビュー] をクリックすると、結果の表が表示されます。値列の数値は、摂氏度数です。

クリーンアップ

このチュートリアル用に作成した個々のリソースを削除します。

AWS S3 バケット内の holidays.csv ファイルを削除します。
作成した AWS S3 バケットを削除します。
このチュートリアル用に作成した Cloud Storage バケットを削除します。
BigQuery データセットを削除します。
Cloud Composer 環境を削除します（環境のバケットを手動で削除します）。