Airflow dbt ELT Mastery: Fix the 200-Model Timeout

Your dbt run compiled 200 models in one task and died at minute 61. Logs showed a wall of text and no clear culprit.

You'll split that monolith into mapped groups with sane timeouts. Heavy models get room, light models fly.

We cover one-task versus mapped patterns, Airflow 3 dbt operators, and test gates that stop bad data. Timeouts stop being scary.

Split the run. See the failure.

Why dbt Belongs Inside Airflow

dbt transforms with tests and docs, Airflow schedules with retries and alerts. Together they give ELT with lineage and SLAs.

Airflow decides when and in what order models build; dbt decides how each model compiles. That split keeps orchestration and SQL cleanly separated.

The Two Patterns: One Task vs Mapped Models

One task runs dbt build for everything: simple, but one timeout and one log for 200 models. Mapped tasks run one group each with isolated retries.

Start mapped once you pass 20 models or 15 minutes runtime. The debugging payoff dwarfs the extra wiring within a week.

Cosmos is the community standard here: DbtTaskGroup drops a dbt project into a normal DAG as a task group while DbtDag turns a whole project into its own DAG, each model becoming a task with lineage in the UI. Dependencies between dbt models become Airflow dependencies automatically, and dbt tests attached to a model run right after it, so failures point at the model, not a 200-model hairball. This beats the BashOperator fallback, where one dbt build means one log, absolute retries, and near-zero observability.

from airflow.decorators import dag, task
from datetime import datetime

MODEL_GROUPS = ["orders", "customers", "inventory", "payments", "marketing", "finance", "events", "core"]

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["dbt"])
def marts_build():
    @task(pool="analytics_light", execution_timeout=600)
    def build_group(group: str):
        import subprocess
        subprocess.run(["dbt", "build", "--select", f"tag:{group}"], check=True)
        return {"group": group}

    @task
    def run_tests():
        import subprocess
        subprocess.run(["dbt", "test"], check=True)
        return {"tested": True}

    built = build_group.expand(group=MODEL_GROUPS)
    built >> run_tests()

marts_build()
# from cosmos import DbtTaskGroup, DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
# DbtTaskGroup(group_id=..., project_config=..., profile_config=...)  # models -> tasks
# profile_mapping=PostgresUserPasswordProfileMapping(conn_id="db_conn")  # no profiles.yml

Airflow 3.0 dbt Operators

Airflow 3 ships first-class dbt operators including DbtRun paths for managed runs. They add templating and connection handling over raw subprocess.

Use operators for standard build and test steps, subprocess Tasks for custom selectors. Either way, keep one group per task for clear lineage.

Wire Cosmos with three configs: ProjectConfig points at the project dir, ProfileConfig maps an Airflow connection (PostgresUserPasswordProfileMapping and friends) so no profiles.yml ships with code, and ExecutionConfig pins the dbt binary, often a dbt_venv virtualenv to dodge dependency clashes. Inject Airflow values with operator_args vars like '{"my_name": "{{ params.my_name }}"}', set retries to at least 2 on model tasks, and bump dagbag_import_timeout when big projects trip DagBag import limits. Can't co-locate the project? Parse a manifest.json instead, or run containerized execution modes; the watcher mode can cut large-project runtimes dramatically.

Task Mapping Over Model Groups

expand() fans one task definition over model groups with per-map timeouts. Heavy aggregates get 45 minutes while dimensions get 10.

Downstream joins with map_index to aggregate results. A failed map retries alone instead of rerunning all 200 models.

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

@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False, tags=["dbt"])
def marts_mapped():
    @task(pool="analytics_heavy", execution_timeout=timedelta(minutes=45), retries=2)
    def build_heavy(model: str):
        import subprocess
        subprocess.run(["dbt", "build", "--select", model], check=True)
        return {"model": model}

    heavy = build_heavy.expand(model=["orders_rollup", "revenue_daily", "cohort_monthly"])

    @task(trigger_rule="none_failed")
    def join_results(results: list):
        return {"built": len(results)}

    join_results(heavy)

marts_mapped()

Timeouts and Long Runs

Set execution_timeout per group from Snowflake history p95 plus 50% headroom. Global DAG timeouts punish heterogeneous models.

Cap heavy groups with small Pools so two monsters never run together. Long runs need isolation more than they need patience.

dbt Tests as Airflow Quality Gates

Run dbt test as a downstream task that blocks publish. Failed uniqueness or not-null tests stop marts from shipping bad rows.

Keep tests in the same mapped grouping so failures point at the model. Green tests mean the publish task may run.

Put dbt test tasks directly after their models inside the same group and let Airflow retries plus error notifications do the paging. Surround the project with Airflow sensors or data-aware scheduling so models build when upstream events land, and add your own SQL quality operators beside dbt tests for warehouse-specific rules. Cosmos can even generate and host dbt docs from the same DAG for one-click lineage reviews.