dbt with Airflow

Apache Airflow is the most widely used workflow orchestration tool in data engineering. It lets you define complex pipelines as Python code, schedule them, monitor runs, and set dependencies between tasks. Combining Airflow with dbt gives you fine-grained control over when dbt runs, how it integrates with other pipeline steps, and how failures are handled and retried.

Why Use Airflow with dbt

Airflow alone:          moves and schedules data tasks
dbt alone:              transforms data inside the warehouse
Airflow + dbt:          orchestrates the full pipeline end to end

Full pipeline example:
  [Airflow] Extract from Salesforce API
      |
  [Airflow] Load raw data to Snowflake
      |
  [Airflow] Trigger: dbt source freshness check
      |
  [Airflow] Trigger: dbt build (run + test)
      |
  [Airflow] Trigger: update Tableau extract
      |
  [Airflow] Send success notification to Slack

Three Integration Approaches

Approach 1: BashOperator (Simplest)

Run dbt CLI commands as bash scripts inside Airflow tasks:

from airflow.operators.bash import BashOperator

dbt_run = BashOperator(
    task_id='dbt_run',
    bash_command='cd /opt/dbt_project && dbt build --target prod',
    env={
        'SNOWFLAKE_PASSWORD': '{{ var("snowflake_password") }}',
        'DBT_PROFILES_DIR': '/opt/dbt_project'
    }
)

Approach 2: astronomer-cosmos (Recommended)

The astronomer-cosmos package converts your entire dbt project into individual Airflow tasks — one task per dbt model. This gives you model-level visibility, retry control, and dependency visualization inside Airflow.

from cosmos import DbtDag, ProjectConfig, ProfileConfig, RenderConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

dbt_dag = DbtDag(
    dag_id="dbt_project",
    schedule_interval="0 6 * * *",
    project_config=ProjectConfig("/opt/dbt_project"),
    profile_config=ProfileConfig(
        profile_name="my_project",
        target_name="prod",
        profile_mapping=SnowflakeUserPasswordProfileMapping(
            conn_id="snowflake_default",
            profile_args={"schema": "analytics"}
        )
    ),
    render_config=RenderConfig(
        select=["tag:daily"]   ← only render daily-tagged models
    )
)

In the Airflow UI, each dbt model appears as its own task box. Failed models show in red. You can retry a single model without rerunning the whole pipeline.

Approach 3: dbt Cloud API

Trigger a dbt Cloud job from Airflow using the HTTP operator:

from airflow.providers.http.operators.http import SimpleHttpOperator

trigger_dbt_cloud = SimpleHttpOperator(
    task_id='trigger_dbt_cloud_job',
    http_conn_id='dbt_cloud_api',
    endpoint=f'/api/v2/accounts/{ACCOUNT_ID}/jobs/{JOB_ID}/run/',
    method='POST',
    headers={"Authorization": f"Token {DBT_API_TOKEN}"},
    data='{"cause": "Triggered by Airflow"}'
)

Full DAG Example with BashOperator

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime, timedelta

DBT_DIR = "/opt/dbt_project"
DBT_CMD = f"cd {DBT_DIR} && dbt"

default_args = {
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    dag_id='dbt_daily_pipeline',
    schedule_interval='0 6 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False,
    default_args=default_args
) as dag:

    install_packages = BashOperator(
        task_id='dbt_deps',
        bash_command=f'{DBT_CMD} deps'
    )

    check_freshness = BashOperator(
        task_id='dbt_source_freshness',
        bash_command=f'{DBT_CMD} source freshness'
    )

    run_staging = BashOperator(
        task_id='dbt_run_staging',
        bash_command=f'{DBT_CMD} build --select tag:staging'
    )

    run_marts = BashOperator(
        task_id='dbt_run_marts',
        bash_command=f'{DBT_CMD} build --select tag:mart'
    )

    install_packages >> check_freshness >> run_staging >> run_marts

Passing Variables from Airflow to dbt

# Pass Airflow execution date as dbt variable
run_dbt = BashOperator(
    task_id='dbt_run',
    bash_command=(
        'dbt run '
        '--vars \'{"run_date": "{{ ds }}"}\' '  ← ds = execution date YYYY-MM-DD
        '--target prod'
    )
)

Retry Strategy

default_args = {
    'retries': 3,                          ← retry failed tasks 3 times
    'retry_delay': timedelta(minutes=10),  ← wait 10 min between retries
    'retry_exponential_backoff': True      ← double wait time each retry
}

When to Use Each Approach

Approach                 Best When
-------------------      ---------
BashOperator             Simple pipelines, quick setup needed
astronomer-cosmos        Model-level visibility needed in Airflow UI
dbt Cloud API trigger    Already using dbt Cloud for scheduling