How to trigger a dbt Cloud job from Airflow

When dbt runs in dbt Cloud, Airflow only has to start the job and wait for the result. The dbt Cloud provider does both with DbtCloudRunJobOperator: it calls the dbt Cloud API, follows the run until it finishes and fails the task if the job fails.

Copy and paste

1. Install the provider:

pip install apache-airflow-providers-dbt-cloud

2. Create the connection. Account ID in login, a service token in password, and your account’s host in host:

export AIRFLOW_CONN_DBT_CLOUD_DEFAULT='{"conn_type": "dbt_cloud", "login": "12345", "password": "<service token>", "host": "cloud.getdbt.com"}'

cloud.getdbt.com is the default host. Accounts on another region or on a dedicated cell use the host shown in their dbt Cloud account settings.

3. Run the job:

from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator

run_dbt_job = DbtCloudRunJobOperator(
    task_id="run_dbt_job",
    dbt_cloud_conn_id="dbt_cloud_default",
    job_id=123456,
    check_interval=60,
    timeout=2 * 60 * 60,
    deferrable=True,
)

Pass the run date, or replace the job’s commands for this run:

run_dbt_job = DbtCloudRunJobOperator(
    task_id="run_dbt_job",
    job_id=123456,
    steps_override=["dbt build --select tag:daily --vars '{run_date: \"{{ ds }}\"}'"],
    deferrable=True,
)

Find the job by name instead of ID:

run_dbt_job = DbtCloudRunJobOperator(
    task_id="run_dbt_job",
    project_name="Shop",
    environment_name="Production",
    job_name="Daily build",
)

Waiting for the job

• wait_for_termination=True, the default, keeps the task running until the dbt Cloud run finishes. The task fails when the job fails.
• check_interval is how often, in seconds, the operator asks for the status. The default is 60.
• timeout limits how long the operator waits, in seconds; the default is seven days. When it expires the task fails, but the job keeps running in dbt Cloud.
• deferrable=True hands the waiting to the triggerer, so the task does not hold a worker slot for the length of the job. The triggerer must be running, as it is with airflow standalone.

Other options

• steps_override, schema_override, trigger_reason and additional_run_config are templated, so they can use {{ ds }} and the other variables.
• reuse_existing_run=True attaches to a run of the same job that is still in progress instead of starting a new one.
• retry_from_failure=True: when the last run of the job failed, the operator calls dbt Cloud’s rerun endpoint, which restarts that run from the point of failure instead of from the beginning.
• The operator returns the dbt Cloud run ID, so {{ ti.xcom_pull(task_ids='run_dbt_job') }} gives it to later tasks. It also stores the run URL under the XCom key job_run_url.
• With wait_for_termination=False the task only starts the job. A DbtCloudJobRunSensor from airflow.providers.dbt.cloud.sensors.dbt can wait for that run ID later in the DAG.
• Without Airflow, the same API calls work from any script; see how to trigger a dbt Cloud job from the API.

Checked against the code of the dbt Cloud provider 4.9 on Airflow 3.3.2; not run against a live dbt Cloud account.