BigQuery Data Transfer Service で GCS のパスを動的に指定する

こんにちは、川田です。

今回は BigQuery Data Transfer Service で、取込対象となる Google Cloud Storage 上オブジェクト・パスを動的に指定する方法を確認します。

結論

ランタイム・パラメーターの機能を利用します。

Cloud Storage の転送でのランタイム パラメータ

例えば、以下のようなパラメーターで転送を設定した場合（key data_path_template の gcs パスに注目）。

$ params=$(cat <<-EOF
{
    "destination_table_name_template": "sample",
    "data_path_template": "gs://raw-bucket-asia-northeast1-cm-kawata/input/{run_time|\"%Y-%m-%d\"}/{run_time|\"%H\"}/input.csv",
    "write_disposition": "APPEND",
    "file_format": "CSV",
    "allow_jagged_rows": "true"
}
EOF
)
$ bq mk \
--transfer_config  \
--project_id=<PROJECT-ID> \
--data_source=google_cloud_storage \
--display_name=job-dynamic-object-path \
--target_dataset=demo \
--service_account_name=sa-bq-transfer-job@<PROJECT-ID>.iam.gserviceaccount.com \
--params="${params}"

転送の実行時刻が 2024-04-03 03:00:00 (UTC) の場合、以下のような GCS パスとして評価されます。

gs://raw-bucket-asia-northeast1-cm-kawata/input/2024-04-03/03/input.csv

以降で、実際に設定を行い動作を確認します。

事前準備

事前の準備を行います。

BigQuery 側でロード先となるテーブルを作成

データのロード先となるデータセットを作成します。今回は demo という名前で作成しました。

$ bq mk --location=asia-northeast1 \
--dataset \
--max_time_travel_hours=48 \
--storage_billing_model=PHYSICAL \
<PROJECT-ID>:demo

データのロード先となるテーブルを作成します。今回は sample という名前で作成しました。

$ bq mk \
--table \
<PROJECT-ID>:demo.sample \
col1:STRING,col2:STRING

取込対象となるファイルを用意

ファイルを配置する GCS のバケットを作成します。

$ gcloud storage buckets create gs://raw-bucket-asia-northeast1-cm-kawata  \
--project <PROJECT-ID> --location asia-northeast1

取込ファイルとなる CSV ファイルを、作成したバケット上に作成します。

$ cat <<EOF | gcloud storage cp - gs://raw-bucket-asia-northeast1-cm-kawata/input/2024-04-03/03/input.csv
AAA,AAA
BBB,BBB
EOF
$
$ gcloud storage cat gs://raw-bucket-asia-northeast1-cm-kawata/input/2024-04-03/03/input.csv
AAA,AAA
BBB,BBB

作成先のパスは下記となります。

gs://raw-bucket-asia-northeast1-cm-kawata/input/2024-04-03/03/input.csv

BigQuery Data Transfer Service 向けのサービスアカウントを作成

BigQuery Data Transfer Service で利用するサービスアカウントを作成します。

$ gcloud --project <PROJECT-ID> iam service-accounts create sa-bq-transfer-job \
--description="sa-bq-transfer-job" \
--display-name="sa-bq-transfer-job"

作成したサービスアカウントに、roles/bigquery.admin と roles/storage.objectUser の事前定義ロールを付与しておきます。

$ gcloud projects add-iam-policy-binding <PROJECT-ID> \
--member="serviceAccount:sa-bq-transfer-job@<PROJECT-ID>.iam.gserviceaccount.com" \
--role="roles/bigquery.admin"
$ gcloud projects add-iam-policy-binding <PROJECT-ID> \
--member="serviceAccount:sa-bq-transfer-job@<PROJECT-ID>.iam.gserviceaccount.com" \
--role="roles/storage.objectUser"

必要な権限

BigQuery Data Transfer Service の転送の設定

それでは、実際に転送を設定していきます。

動的に GCS のパスを指定するには、key data_path_template の値を、ランタイムパラメーターを利用して指定します。

パラメーターとして利用できるものに、以下の２つが用意されています。

パラメーターには、テンプレートシンタックスが用意されており、以下のように値を操作することができます。

使用上の注意があります。

• run_time、offset、time 形式の間に空白文字は使用できません。
• 文字列にリテラル中かっこを含めるには、'\{' and '\}' としてエスケープできます。
• time_format に "YYYY|MM|DD" などのリテラル引用符や縦線を含めるには、'\"' や '\|' のフォーマット文字列でエスケープします。

より詳細な情報は、公式ドキュメントを参照ください。

Cloud Storage の転送でのランタイム パラメータ

転送を設定するコマンドは以下となります（key data_path_template の gcs パスに注目）。

$ params=$(cat <<-EOF
{
    "destination_table_name_template": "sample",
    "data_path_template": "gs://raw-bucket-asia-northeast1-cm-kawata/input/{run_time|\"%Y-%m-%d\"}/{run_time|\"%H\"}/input.csv",
    "write_disposition": "APPEND",
    "file_format": "CSV",
    "allow_jagged_rows": "true"
}
EOF
)
$ bq mk \
--transfer_config  \
--project_id=<PROJECT-ID> \
--data_source=google_cloud_storage \
--display_name=job-dynamic-object-path \
--target_dataset=demo \
--service_account_name=sa-bq-transfer-job@<PROJECT-ID>.iam.gserviceaccount.com \
--params="${params}"

上記の data_path_template の値では、転送の実行時刻が 2024-04-03 03:00:00 (UTC) の場合、以下のような GCS パスとして評価されます。

gs://raw-bucket-asia-northeast1-cm-kawata/input/2024-04-03/03/input.csv

コマンドラインを実行したタイミングで、転送が作成され、実際の転送処理が開始されます。コマンドラインから作成の場合、作成時刻を基準として 24 時間間隔のスケジュールが自動的に設定されます。

Caution: When you create a Cloud Storage transfer using the command-line tool, the transfer configuration is set up using the default value for Schedule (every 24 hours).

Set up a Cloud Storage transfer

確認

転送の処理結果を確認します。

ランタイムパラメーターが正しく評価されて、テーブルへのロードが成功しています。

その他補足

以下、補足となります。

• data_path_template での GCS パスの指定では、ランタイムパラメーターのほか、ワイルドカード（*）も利用できます
• ランタイムパラメーターでの値の指定は、ロード先の BigQuery テーブルを登録する Key destination_table_name_template でも利用できます
• AWS S3 上のパスを指定する場合にも、ランタイムパラメーターを利用できます