dbt Cloud APIのTrigger Job Runのsteps_overrideを活用してdbtコマンドを制御してみた

かわばたです。

dbt CloudのジョブをAPIを介して実行する方法を下記で試しました。
https://dev.classmethod.jp/articles/dbt-cloud-api/

今回は表題のとおり、Trigger Job Runの詳細を確認しつつ、オプションパラメータのsteps_overrideを活用してdbtコマンドをAPI経由で制御していきます。

【公式ドキュメント】
https://docs.getdbt.com/dbt-cloud/api-v2#/operations/Trigger Job Run

対象読者

• dbt Cloud APIでdbtコマンド制御に興味のある方

検証環境と事前準備

検証環境

• SnowflakeトライアルアカウントEnterprise版
• dbt CloudアカウントEnterprise版

事前準備

dbtのサンプルデータであるjaffle_shopを活用しています。

今回行わないこと

https://dev.classmethod.jp/articles/dbt-cloud-api/

• 上記で設定しているAPIの設定
• Airflowのセットアップ

概要

このエンドポイントを使用して、ジョブの実行を開始（キックオフ）します。
このエンドポイントが成功応答を返すと、新しい実行がアカウントのキュー（待ち行列）に追加されます。
ユーザーは、実行が完了するまで「Get run (実行状況の取得)」エンドポイントをポーリング（定期的に問い合わせ）して、実行状況を確認できます。
実行が完了した後、ユーザーは「Get run artifact (実行成果物の取得)」エンドポイントを使用して、その実行によって生成されたアーティファクト（成果物）をダウンロードできます。

上記ドキュメントからの引用となりますが、簡単に説明すると「ジョブの実行を開始する」ための、具体的なAPIエンドポイントとなります。

Path Parameters

下記パラメータは必須の項目となります。

• account_id
• job_id

Body

下記パラメータがオプションパラメータとなります。

今回はよく使用されそうなsteps_overrideを試していきます。

実際に試してみた

cURLで実行

ターミナルで実行していきます。
以下のコマンドを実行する際に、YOUR_ACCOUNT_ID,YOUR_JOB_ID,YOUR_API_TOKENを自身の値に置き換えてください。

curl -X POST \
  -H "Authorization: Token {YOUR_API_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"cause": "Triggered by external orchestrator for daily finance models.", "steps_override": ["dbt seed --select raw_products", "dbt run --select products"]}' \
  "https://cloud.getdbt.com/api/v2/accounts/{ACCOUNT_ID}/jobs/{JOB_ID}/run/"

上記のように、dbt seed --select raw_products、dbt run --select productsのみ実行されていることが分かります。
※元の設定は下記となります。

steps_overrideの設定により、元々の設定が上書きされていることが分かります。

schema_overrideも試してみます。

curl -X POST \
  -H "Authorization: Token {YOUR_API_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"cause": "Triggered by external orchestrator for daily finance models.",\
 "schema_override":"test_override", \
 "steps_override": ["dbt seed --select raw_products", "dbt run --select +products"]}' \
  "https://cloud.getdbt.com/api/v2/accounts/{ACCOUNT_ID}/jobs/{JOB_ID}/run/"

実行結果は下記のとおりです。

連携しているSnowflakeも確認します。
test_overrideスキーマに新しく連携されていることが分かります。

Airflowで実装

先ほどのAPIを活用するケースとして、オーケストレーションツールとの連携があります。
代表的なAirflowでどのように実装できるか簡単な例で確認していきます。

Airflowは、dbt Cloud APIとの連携を簡素化するための公式プロバイダーパッケージapache-airflow-providers-dbt-cloudを提供しています。
https://airflow.apache.org/docs/apache-airflow-providers-dbt-cloud/stable/_api/airflow/providers/dbt/cloud/operators/dbt/index.html

Airflow接続のセットアップ

1. Airflow UIにログインし、上部メニューのAdmin > Connectionsを開きます。

2. +Add Connectionボタンをクリックして、新しい接続を追加します。

3. 以下のようなポップアップ画面が出るので、

• Connection ID：任意の名称を記載してください
• Connection Type：dbt Cloudを選択してください
• Account ID：dbt CloudのAccount IDを入力してください
• API Token：dbt CloudのAPI Tokenを入力してください

4. すべて入力ができましたら、以下のように登録されます。

dagの作成

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.models.dag import DAG
import pendulum

with DAG(
    dag_id="dbt_cloud_specific_model_run",
    start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
    catchup=False,
    schedule=None,
    tags=["dbt", "dbt_cloud"],
) as dag:
    trigger_specific_model_run = DbtCloudRunJobOperator(
        task_id="trigger_dbt_cloud_specific_model",
        dbt_cloud_conn_id="Connection IDの値を入力",  # Airflow Connectionで設定した接続ID
        job_id=12345,  # dbt Cloud上の汎用的なジョブID
        schema_override="test_override",
        steps_override=["dbt seed --select raw_stores","dbt run --select +locations --warn-error"],
        check_interval=10,
        timeout=300,
    )

以下のように実行することができました。

ジョブが正常に完了していることが確認できました。

連携先のSnowflakeにもしっかりと反映されています。

【小ネタ】同じジョブで連続で実行するとどうなるか

下記のような形で、先に実行したものから実行され、後に実行したものがキューとして待機する形になります。

同じジョブで実行するか、ジョブを分けて設計するかはユースケースによって異なりそうです。

最後に

Trigger Job Runのオプションパラメータを使用してみましたが、overrideのオプションは活用できそうでした。

• 特定のモデルのみを実行したい、回し直したいケース
• CI/CDパイプラインでの変更差分実行するケース(dbt run --select state:modified+)
• 異なるスキーマに出力し検証したいケース

また、オーケストレーションツールを活用することで幅が広がりそうなので、試していきたいです。
この記事が何かの参考になれば幸いです！