Senior Data Engineer

Incremental dbt + Airflow Pattern:

Step 1 — Get changed models from git diff:

# dags/dbt_incremental_dag.py
from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator
import subprocess

@task
def get_changed_models() -> str:
    result = subprocess.run(
        ['git', 'diff', '--name-only', 'HEAD~1', 'HEAD', '--', 'models/'],
        capture_output=True, text=True, cwd='/opt/dbt/project'
    )
    changed_files = result.stdout.strip().split('\n')
    # Convert file paths to dbt model selectors
    models = [f.replace('models/', '').replace('.sql', '').replace('/', '.') for f in changed_files if f.endswith('.sql')]
    return ' '.join(f'+{m}' for m in models)  # + prefix includes downstream

@dag(schedule='@hourly', catchup=False)
def dbt_incremental():
    changed = get_changed_models()
    BashOperator(
        task_id='dbt_run',
        bash_command=f'dbt run --select {{ changed }} --profiles-dir /opt/dbt',
    )

The +model_name selector runs the model AND all downstream dependents — this is what you want for correctness.

For a more robust setup, use the dbt Cloud API or dbt-airflow provider:

pip install astronomer-cosmos  # NASA's dbt-Airflow integration

Cosmos parses the dbt manifest and creates individual Airflow tasks per model — giving you granular retry, skip, and observability.

Expected result: 4h full refresh → 20-40min for a typical PR touching 10-20 models.