Airflow + dbt: The 200-Model Run That Hit the Timeout

dbt is the transformation layer. Airflow is the orchestrator. Together they're the default ELT stack of the modern warehouse, and the integration is where most teams stumble.

The failure is predictable: a team schedules a single dbt run task with a 60-minute timeout. The model graph grows to 200 models. One heavy model hits 55 minutes, the task times out, the whole pipeline fails, and the alert lands at 3 AM.

Here are the two integration patterns, the Airflow 3.0 operators, and the timeout math that decides which one you need.

1. Why dbt Belongs Inside Airflow

dbt owns the transformation: models, tests, documentation, and the dependency graph between them. Airflow owns everything around the transformation: when it runs, what runs first, what happens when it fails, and who gets paged.

The common mistake is letting one tool do both. Some teams try to schedule dbt from cron alone and lose retries, state, and visibility. Others try to re-implement the model graph in Airflow with one task per model written by hand — duplicating what dbt already knows.

The clean split: dbt describes the graph, Airflow executes and watches it.

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta

with DAG(
    dag_id="elt_warehouse",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
) as dag:

    @task
    def run_dbt():
        # dbt owns the transform graph; Airflow owns the schedule
        from dbt.cli.main import dbtRunner
        dbtRunner().invoke(["run", "--profile", "prod"])

    run_dbt()

2. The Two Integration Patterns

There are two production patterns for wiring dbt into Airflow.

Pattern one: a single task that runs the whole project. Simple, one task, one log. It's the right call for small projects (under ~50 models) where the whole run completes well inside your timeout.

Pattern two: task mapping over model groups. You enumerate model groups in a config file, then expand one task per group. Failures localize to the exact group, and parallel groups cut total runtime.

Pattern two is what the incident article needed — but it costs a bit of setup. Here's the trade-off table.

dbt's own guidance threads the needle: bundle as many models per command as is practical — dbt orders them by dependency and parallelizes where it can, and a failed run can resume from the point of failure (turn on retry_from_failure for dbt Cloud jobs) — but group by run frequency or department rather than one job per node. The bug isn't bundling; it's bundling everything behind a single Airflow task timeout.

3. Airflow 3.0 dbt Operators

Airflow 3.0 ships first-party dbt integration as part of the standard library, and the old community plugin pattern (airflow_dbt with DbtRunOperator) is on its way out.

The new operators accept the same arguments you'd pass to the dbt CLI: project_dir, profile_name, select, and flags. You get templated SQL via Jinja — {{ ds }} works inside dbt run arguments — and full integration with Airflow's retry and alerting model.

The practical upgrade path: keep your dbt project untouched, swap the custom plugin calls for the first-party operators, and delete the plugin dependency from requirements.

Two official paths cover the rest of the integration space. For managed dbt Cloud jobs, the apache-airflow-providers-dbt-cloud package ships DbtCloudRunJobOperator (trigger a job and wait, or set wait_for_termination=False and pair it with DbtCloudJobRunSensor), DbtCloudGetJobRunArtifactOperator (download manifest.json, catalog.json, and run_results.json from the run's target/), plus job-listing operators — the account is bound via a dbt_cloud connection or an explicit account_id. For dbt Core, the Astronomer-maintained Cosmos package is the ecosystem standard: DbtTaskGroup or DbtDag turns the dbt project into real Airflow tasks, resolves profiles from Airflow connections, and can run dbt in its own virtual environment to avoid dependency conflicts.

from airflow import DAG
from airflow.operators.dbt import DbtRunAirflowOperator, DbtTestAirflowOperator
from datetime import datetime

with DAG(
    dag_id="elt_warehouse",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    run_staging = DbtRunAirflowOperator(
        task_id="run_staging",
        project_dir="/opt/airflow/dbt/sales",
        select="staging_*",
    )

    test_marts = DbtTestAirflowOperator(
        task_id="test_marts",
        project_dir="/opt/airflow/dbt/sales",
        select="marts_*",
    )

    run_staging >> test_marts

4. Task Mapping Over Model Groups

Task mapping is the pattern that fixes the timeout incident. You define the model groups once, then expand() a dbt task over them.

Each group becomes its own task instance — visible in the grid view, retryable in isolation, and monitored separately. The slow model's group gets a dedicated timeout, the fast groups finish in minutes.

What about map_index? Every mapped task instance gets an index. Downstream tasks that depend on the mapped run receive the full set of indices, so joins and aggregates over the group results stay correct.

from airflow import DAG
from airflow.decorators import task
from datetime import datetime

MODEL_GROUPS = [
    {"name": "staging", "select": "staging_*", "timeout_min": 15},
    {"name": "marts", "select": "marts_*", "timeout_min": 30},
    {"name": "monthly_rebuild", "select": "monthly_rebuild", "timeout_min": 90},
]

with DAG(
    dag_id="elt_warehouse",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    @task
    def run_dbt_group(select: str, timeout_min: int):
        from dbt.cli.main import dbtRunner
        dbtRunner().invoke(["run", "--select", select, "--profile", "prod"])

    run_dbt_group.expand(
        select=[g["select"] for g in MODEL_GROUPS],
        timeout_min=[g["timeout_min"] for g in MODEL_GROUPS],
    )

5. dbt Tests as Airflow Quality Gates

dbt tests are only useful if a failure stops the right thing. The production pattern: dbt tests run as their own task after the model run, so a test failure doesn't re-run the load — it blocks promotion to the next stage.

Your quality gate chain looks like this: run models -> run tests -> mark the table promoted. If a test fails, the promotion task never runs, and the alert goes to the data team with the failing test name in the task log.

Use --store-failures so failed test rows land in the warehouse where your team can query them instead of hunting through logs.

from airflow import DAG
from airflow.operators.dbt import DbtRunAirflowOperator, DbtTestAirflowOperator
from airflow.operators.python import PythonOperator
from datetime import datetime

def promote_table():
    print("Promoting marts tables: tests passed, freshness confirmed")

with DAG(
    dag_id="elt_warehouse",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    run_marts = DbtRunAirflowOperator(
        task_id="run_marts", project_dir="/opt/airflow/dbt/sales", select="marts_*"
    )
    test_marts = DbtTestAirflowOperator(
        task_id="test_marts",
        project_dir="/opt/airflow/dbt/sales",
        select="marts_*",
        flags=["--store-failures"],
    )
    promote = PythonOperator(task_id="promote_tables", python_callable=promote_table)

    run_marts >> test_marts >> promote

6. Timeouts, Retries, and the 3 AM Alert

The incident article taught the whole team one lesson: a timeout is a signal, not a knob. Raising the timeout turns a nightly failure into a nightly near-failure.

Set per-task timeouts. The staging group gets 15 minutes, the monthly rebuild gets 90. Retries with exponential backoff apply only to the failing group, so a transient warehouse blip retries the rebuild without touching the fast groups.

The alert policy: on_d_failure callbacks should carry the failing task id and map_index into the notification. The on-call engineer reads "monthly_rebuild (index 2) timed out" instead of "DAG failed".

Two dbt Cloud knobs belong in this section. deferrable=True on the job-run operator hands the polling off to the triggerer, so a long job doesn't pin a worker slot. And retry_from_failure=True resumes a failed job from the point of failure instead of starting the whole run over — the managed equivalent of isolating the slow model.