Cloud Run Job で dbt を実行して Managed Airflow からオーケストレーションする手順

はじめに

本記事では、以下の記事で作成される Silver テーブルを入力として、dbt を使って Gold テーブルを作成します。

dbtの実行環境には Cloud Run Job を利用し、dbt プロジェクトを Docker コンテナとして実行できるようにします。さらに、Managed Service for Apache Airflow（旧Cloud Composer）から Cloud Run Job を呼び出し、Airflow から dbt の処理を実行できるところまで確認します。

最終的な構成は以下のイメージです。

BigQuery
  └─ Silver テーブル
       ↓
Cloud Run Job
  └─ dbt
       ↓
BigQuery
  └─ Gold テーブル

Managed Airflow
  └─ Airflow DAG
       ↓
Cloud Run Job を実行
       ↓
dbt を実行

Cloud Shellを開く

本記事では基本的にCloud Shellからコマンドを実行します。

Google Cloud コンソールの右上のアイコンから Cloud Shell を開きます。

以降のコマンドで繰り返し使用するプロジェクト ID やリージョン、リソース名をここで環境変数として定義しておきます。

export PROJECT_ID="<PROJECT_ID>"
export REGION="asia-northeast1"

export DATASET_ID="medallion_demo"

export REPOSITORY="dbt-images"
export IMAGE_NAME="dbt-gold"
export JOB_NAME="dbt-product-count-by-family-job"

export DBT_SERVICE_ACCOUNT="dbt-runner"
export COMPOSER_SERVICE_ACCOUNT="composer-runner"
export COMPOSER_ENV="dbt-gold-airflow"

gcloud config set project "${PROJECT_ID}"

echo "${PROJECT_ID}"
echo "${REGION}"

必要なAPIを有効化する

Cloud Run、Cloud Build、Artifact Registry、BigQuery、Managed Airflow など、今回利用する Google Cloud サービスの API を有効化します。

gcloud services enable \
  run.googleapis.com \
  artifactregistry.googleapis.com \
  cloudbuild.googleapis.com \
  bigquery.googleapis.com \
  iam.googleapis.com \
  composer.googleapis.com \
  logging.googleapis.com

各 API の用途は以下となります。

API	用途
Cloud Run API	dbtコンテナの実行
Artifact Registry API	コンテナイメージの保存
Cloud Build API	Dockerイメージのビルド
BigQuery API	dbtからのSQL実行
IAM API	サービスアカウントの作成
Composer API	Managed Airflow環境の作成
Logging API	実行ログの保存

BigQuery データセットを作成する

dbt の入力元となる Silver テーブルと、Gold テーブルの出力先となる BigQuery データセットを準備します。今回は以前の記事で作成した データセットと Silverテーブルを利用します。

以下のコマンドでデータセットにあるテーブルの一覧を確認することができます。

bq ls "${DATASET_ID}"

Cloud Run 用サービスアカウントを作成する

Cloud Run Job から BigQuery へ安全にアクセスするため、dbt 実行専用のサービスアカウントを作成します。Cloud Run Job には、このサービスアカウントを実行時の ID として設定します。

gcloud iam service-accounts create "${DBT_SERVICE_ACCOUNT}" \
  --display-name="dbt Cloud Run runner" \
  --description="Cloud Run Jobからdbtを実行するサービスアカウント"

export DBT_SA_EMAIL="${DBT_SERVICE_ACCOUNT}@${PROJECT_ID}.iam.gserviceaccount.com"
echo "${DBT_SA_EMAIL}"

作成したサービスアカウントは以下の UI で確認できます。

サービスアカウントに BigQuery 権限を付ける

作成したサービスアカウントから BigQuery のクエリ実行や Gold テーブルの作成ができるように、必要な IAM ロールを付与します。

ロール	用途
BigQuery Job User	BigQueryクエリジョブを実行
BigQuery Data Editor	テーブルの作成・更新・削除

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${DBT_SA_EMAIL}" \
  --role="roles/bigquery.jobUser"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${DBT_SA_EMAIL}" \
  --role="roles/bigquery.dataEditor"

権限付与の状況は以下の UI で確認できます。

dbt プロジェクトを作成する

Cloud Run Job で実行する dbt プロジェクトを作成します。BigQuery への接続設定、Silver テーブルの Source 定義、Gold テーブルを作成する SQL モデルを順番に用意します。

まずは、Cloud Shell で作業用ディレクトリを作成します。

mkdir -p ~/dbt-gold/models
cd ~/dbt-gold

最終的には、次のファイル構成になります。

dbt-dbt-gold/
├── Dockerfile
├── requirements.txt
├── dbt_project.yml
├── profiles.yml
└── models/
    └── sources.yml
    └── product_count_by_family.sql

requirements.txt を作成する

Docker コンテナ内で dbt から BigQuery を利用できるように、必要な Python パッケージを定義します。

今回は BigQuery 用の dbt アダプターである dbt-bigquery をインストールします。

cat > requirements.txt <<'EOF'
dbt-bigquery
EOF

dbt_project.yml を作成する

dbt プロジェクト名やモデルの保存場所、マテリアライズ方法など、プロジェクト全体の基本設定を定義します。

cat > dbt_project.yml <<'EOF'
name: dbt_gold
version: "1.0.0"
config-version: 2

profile: dbt_gold

model-paths:
  - models

models:
  dbt_gold:
    +materialized: table
EOF

設定	意味
name	dbtプロジェクト名
profile	使用する接続設定名
model-paths	SQLモデルを保存するフォルダ
materialized: table	結果をテーブルとして作成

profiles.yml を作成する

dbt から BigQuery へ接続するためのプロファイルを定義します。プロジェクトIDやデータセット名など、環境によって変わる値はCloud Run Jobの環境変数から取得する構成にします。

method: oauth と設定することで、Cloud Run Job に割り当てたサービスアカウントの認証情報が使用されます。

cat > profiles.yml <<'EOF'
dbt_gold:
  target: prod

  outputs:
    prod:
      type: bigquery
      method: oauth
      project: "{{ env_var('DBT_PROJECT_ID') }}"
      dataset: "{{ env_var('DBT_DATASET') }}"
      location: "{{ env_var('DBT_LOCATION') }}"
      threads: 2
EOF

dbt モデルを作成する

入力元となる Silver テーブルを dbt Source として定義し、そのテーブルを集計して Gold テーブルを作成するSQLモデルを作成します。

cat > models/sources.yml <<'EOF'
version: 2

sources:
  - name: silver
    database: "{{ env_var('DBT_PROJECT_ID') }}"
    schema: "{{ env_var('DBT_DATASET') }}"

    tables:
      - name: product2__silver
EOF

cat > models/product_count_by_family.sql <<'EOF'
SELECT
  Family,
  COUNT(*) AS product_count,
FROM 
  {{ source('silver', 'product2__silver') }}
GROUP BY 
  ALL
EOF

ファイルを確認する

ここまでで作成した dbt プロジェクトのファイルが正しく配置されているか、Cloud Shell エディターから確認します。

Google Cloud コンソールの右上のアイコンからエディターを開きます。

ここまでで作成したファイルがあることを確認できます。

Dockerfileを作成する

作成した dbt プロジェクトを Cloud Run Job で実行できるように、dbt と必要なファイルを含んだ Docker イメージを作成するための Dockerfile を用意します。

cat > Dockerfile <<'EOF'
FROM python:3.11-slim

ENV PYTHONUNBUFFERED=1
ENV DBT_PROFILES_DIR=/app

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip \
    && pip install --no-cache-dir -r requirements.txt

COPY dbt_project.yml .
COPY profiles.yml .
COPY models ./models

ENTRYPOINT ["dbt"]
CMD ["run", "--project-dir", "/app", "--profiles-dir", "/app"]
EOF

コンテナが起動すると、最終行の dbt run が実行され、models フォルダのSQLモデルをBigQueryで実行します。

エディターで作成したファイルを確認できます。

Artifact Registryを作成する

Artifact Registry は、Docker コンテナイメージなどを保存するGoogle Cloud のサービスです。

以下のコマンドで Artifact Registry に Docker 形式のリポジトリを作成します。

gcloud artifacts repositories create "${REPOSITORY}" \
  --repository-format="docker" \
  --location="${REGION}" \
  --description="dbt container images"

以下の UI で作成されたことを確認できます。

dbt コンテナをビルドする

Cloud Build コマンドで、現在のフォルダにある Dockerfile を使ってイメージを作成し、Artifact Registryへ保存します。

gcloud builds submit \
  --tag="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/${IMAGE_NAME}:1.0.0"

ビルドが完了すると、以下の UI で確認できます。

権限不足によるエラー

ビルド時に権限不足による以下のエラーが発生しました。

denied: Permission 'artifactregistry.repositories.uploadArtifacts' denied on resource '//artifactregistry.googleapis.com/projects/~~~/locations/asia-northeast1/repositories/dbt-images' (or it may not exist). Remediate access with this Troubleshooter URL or share it with your administrator - https://console.cloud.google.com/iam-admin/troubleshooter/summary;errorId=~~~ . 
ERROR: failed to push because we ran out of retries. 
ERROR 
ERROR: error pushing image "asia-northeast1-docker.pkg.dev/~~~/dbt-images/dbt-gold:1.0.0": retry budget exhausted (10 attempts): step exited with non-zero status: 1

INFO: The service account running this build projects/~~~/serviceAccounts/[e-mail] does not have permission to write logs to Cloud Logging. To fix this, grant the Logs Writer (roles/logging.logWriter) role to the service account.

Compute Engine のデフォルト サービスアカウントに Artifact Registry への書き込み権限と Cloud Logging の権限を付与をすることで解決しました。

export BUILD_SA="[e-mail]"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${BUILD_SA}" \
  --role="roles/artifactregistry.writer"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${BUILD_SA}" \
  --role="roles/logging.logWriter"

Cloud Run Job を作成する

Artifact Registry に保存した dbt コンテナを実行する Cloud Run Job を作成します。

gcloud run jobs deploy "${JOB_NAME}" \
  --image="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/${IMAGE_NAME}:1.0.0" \
  --region="${REGION}" \
  --service-account="${DBT_SA_EMAIL}" \
  --set-env-vars="DBT_PROJECT_ID=${PROJECT_ID},DBT_DATASET=${DATASET_ID},DBT_LOCATION=${REGION}" \
  --tasks=1 \
  --max-retries=0 \
  --task-timeout=10m

各設定の意味は次のとおりです。

設定	意味
--image	実行するdbtコンテナ
--service-account	BigQueryへ接続するアカウント
--set-env-vars	dbtの接続先情報
--tasks=1	コンテナを1つ実行
--max-retries=0	エラー時に自動再実行しない
--task-timeout=10m	最大実行時間を10分に設定

以下の UI で作成されたジョブを確認できます。

Cloud Run Job を手動実行する

作成したCloud Run Job を単体で実行し、dbt が正常に動作することを確認します。

オーバライドを使用して実行 を選択すると、変数の値を設定して実行することができます。

成功すると履歴タブで以下のように確認できます。

オブザーバビリティ タブでログも確認できます。

Cloud Shell からも以下のコマンドでログを確認することができます。

gcloud run jobs logs read \
  ${JOB_NAME} \
  --project="${PROJECT_ID}" \
  --region="${REGION}" \
  --limit=100

実行結果のテーブルも正常にできていました。

Managed Airflow 用サービスアカウントを作成する

ここからは、Managed Airflow から Cloud Run Job を実行する構成を作成します。

Google Cloud の Managed Airflow は、Apache Airflow を Google Cloud 側で管理するサービスです。Google Cloud コンソールやドキュメントでは、以前の名称である Cloud Composer が表示される場合もあります。Managed Airflow 環境では、DAG と呼ばれる Python ファイルで処理の順序やスケジュールを定義します。

まず、Airflow の DAG から Google Cloud のリソースへアクセスするためのサービスアカウントを用意します。

gcloud iam service-accounts create "${COMPOSER_SERVICE_ACCOUNT}" \
  --display-name="Managed Airflow runner" \
  --description="Airflow DAGを実行するサービスアカウント"

export COMPOSER_SA_EMAIL="${COMPOSER_SERVICE_ACCOUNT}@${PROJECT_ID}.iam.gserviceaccount.com"
echo "${COMPOSER_SA_EMAIL}"

作成されたサービスアカウントは以下の UI で確認できます。

Managed Airflow 用の権限を付与する

Managed Airflow 用のサービスアカウントに、Airflow 環境を動作させるための権限と Cloud Run Job を実行・監視するための権限を付与します。

Jobを起動するだけでなく、Cloud Run の Operation をポーリングして完了を待つ動作をするため、Cloud Run Viewerを付与しています。

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${COMPOSER_SA_EMAIL}" \
  --role="roles/composer.worker"

gcloud run jobs add-iam-policy-binding "${JOB_NAME}" \
  --region="${REGION}" \
  --member="serviceAccount:${COMPOSER_SA_EMAIL}" \
  --role="roles/run.invoker"

gcloud projects add-iam-policy-binding "${PROJECT_ID}" \
  --member="serviceAccount:${COMPOSER_SA_EMAIL}" \
  --role="roles/run.viewer"

付与した結果は以下の UI で確認できます。

Managed Airflow 環境を作成する

Apache Airflow のマネージド実行環境として、Managed Service for Apache Airflow 環境を作成します。今回は検証用途のため、Gen 3の小規模な構成で環境を作成します。

ここでは UI で作成し、設定値は以下としました。

項目	設定値
世代	Managed Airflow Gen 3
名前	dbt-gold-airflow
ロケーション	asia-northeast1
イメージバージョン	デフォルト
環境サイズ	小
サービスアカウント	composer-runner
復元性	標準
ネットワーク	デフォルト

Managed Airflow環境は、存在して動作している期間に応じて課金されるため、検証後は削除することを推奨します。

AirflowのDAGファイルを作成する

Cloud Shell へ戻り、Managed Airflow から Cloud Run Job を呼び出すための DAG を作成します。

mkdir -p ~/airflow-dags
cd ~/airflow-dags

次のコマンドで DAG ファイルを作成します。

cat > run_dbt_cloud_run.py <<'EOF'
from __future__ import annotations

from datetime import datetime, timedelta

from airflow import DAG
from airflow.providers.google.cloud.operators.cloud_run import (
    CloudRunExecuteJobOperator,
)

PROJECT_ID = "<PROJECT_ID>"
REGION = "asia-northeast1"
JOB_NAME = "dbt-product-count-by-family-job"

with DAG(
    dag_id="run_dbt_on_cloud_run",
    description="Cloud Run Jobでdbtを実行する",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
    default_args={
        "retries": 0,
        "retry_delay": timedelta(minutes=1),
    },
    tags=["dbt", "cloud-run"],
) as dag:

    run_dbt = CloudRunExecuteJobOperator(
        task_id="run_dbt",
        project_id=PROJECT_ID,
        region=REGION,
        job_name=JOB_NAME,
    )
EOF

作成されたファイルはエディターで確認できます。

DAG を Managed Airflow へアップロードする

作成した DAG ファイルを Managed Airflow 環境の DAG フォルダへアップロードします。

まずは、作成した DAG ファイルをローカルにダウンロードします。

Managed Airflow の画面で「DAG フォルダを開く」を選択します。

Cloud Storage の /dags フォルダが開くので、DAG ファイルをアップロードします。

ファイルが同期されると Managed Airflow の画面で DAG を確認できます。
※ 同期には多少の遅延があります。

airflow_monitoring は、Managed Airflow が自動で用意する稼働確認用の DAG で、一定間隔で自動実行されるようです。

Airflow から DAG を手動実行する

最後に Airflow から DAG を手動実行し、Airflow → Cloud Run Job → dbt → BigQuery という一連の処理が正常に動作することを確認します。

成功すると Airflow 側では以下のようにログが確認できます。

Cloud Run 側でもログを確認でき、実行ユーザーが Managed Airflow 用サービスアカウントとなっていることが確認できます。

Iceberg テーブルへの実行

BigQuery マネージド Iceberg へは実行できるようです。
https://www.getdbt.com/blog/dbt-supports-apache-iceberg-tables-bigquery

一方、Lakehouse for Apache Iceberg への実行は現時点ではできないようです。
https://github.com/dbt-labs/dbt-adapters/issues/2066