BigQuery で Airflow DAG のオーケストレーションができるようになりました（Preview）

エノカワ

2024.10.26

こんにちは！エノカワです。

BigQuery で Airflow DAG をオーケストレーションする機能がプレビューとなりました。
この機能により、BigQuery で Airflow DAG を表示、トリガー、一時停止することが可能になりました。

BigQuery release notes > October 21, 2024

You can now view, trigger, and pause Airflow DAGs in BigQuery.

今回は、この新機能を実際に試してみましたので、その内容をご紹介します。

環境作成

まずは、Cloud Composer 3 環境を作成します。

Google Cloud コンソールで Cloud Composer のページに移動します。

[環境の作成] をクリックし、表示されるバージョンリストから [Composer 3] を選択します。

bigquery-orchestrate-dags_01

test-composer3という名前で、東京リージョン、最新のイメージバージョンを選択し、サービスアカウントなど他はデフォルトのままで作成します。

bigquery-orchestrate-dags_02

DAG を作成する

次に、Cloud Composer 環境で実行するDAGを作成します。

今回は、Google Cloud Storage (GCS) にあるファイルを BigQuery に取り込むための Airflow DAG を作成しました。

gcs_to_bigquery.py

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

PROJECT_ID = '{プロジェクトID}'
DATASET_ID = 'work'
TABLE_ID = 'sales_data'
BUCKET_NAME = 'cm_enokawa_work'

# DAGの基本設定
default_args = {
    'start_date': days_ago(1),
}

with DAG(
    dag_id='gcs_to_bigquery',
    default_args=default_args,
    schedule_interval=None,  # 手動で実行する
    catchup=False
) as dag:

    # GCS から BigQuery へのデータ取り込み
    gcs_to_bq = GCSToBigQueryOperator(
        task_id='load_gcs_to_bq',
        bucket=BUCKET_NAME,  # GCS バケット名
        source_objects=['sales.csv'],  # GCS 内のファイルパス
        destination_project_dataset_table=f'{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}',  # BigQuery のデータセットとテーブル
        source_format='CSV',  # ソースファイル形式
        write_disposition='WRITE_TRUNCATE',  # 既存データの上書き
        skip_leading_rows=1  # CSV のヘッダー行をスキップ
    )

    gcs_to_bq

この DAG では、Google Cloud Storage に保存された CSV ファイル (sales.csv) を BigQuery のテーブル（sales_data）にインポートしています。

この DAG を Cloud Composer にデプロイし、Airflow UI で DAG gcs_to_bigquery が表示されていることを確認します。

bigquery-orchestrate-dags_04

DAG を表示する

BigQuery コンソールで、オーケストレーションページに移動し、DAG 表示を確認してみましょう。

先ほどデプロイした DAG gcs_to_bigquery だけではなく、liveness DAG airflow_monitoring も表示されていました。
BigQuery に関連する DAG だけではなく、Cloud Composer にデプロイされている DAG すべてが表示対象のようです。

bigquery-orchestrate-dags_05