Airflow + dbt: Run dbt From Airflow With Model-Level Visibility

You want to trigger dbt from Airflow without turning your DAG into a single opaque task. The short answer: use Cosmos if you need model-level tasks in the Airflow UI, use the dbt Cloud API if you’re all-in on Cloud and want managed execution, use Bash/Docker for the simplest dbt Core run when a single task is fine, and use KubernetesPodOperator for reproducible containerized runs at scale. Each path trades off visibility, retry granularity, dependency awareness, and developer experience. Below are the real failure modes, selector-based partial runs that actually work, how to pass state for deferral, and code to surface per-model failures in Airflow. Pick the option that fits your team’s constraints, not someone else’s slide.

The options at a glance: visibility vs effort

Four viable ways to run dbt from Apache Airflow show up in production. They differ most on observability, retries, dependency awareness, and how they feel to develop against a real dbt project.

Option	Model visibility	Retry granularity	Dependency awareness	Dev experience
BashOperator (dbt Core CLI)	Low by default	Run-level	Minimal (inside dbt only)	Simple, fast to ship
dbt Cloud API (Operator/Sensor)	Medium (job-level)	Job-level; model retries inside Cloud	Good (Cloud artifacts)	Great if you already use Cloud
Cosmos (generate tasks)	High (per-model tasks)	Model-level	Excellent (Airflow knows graph)	Best DX for dbt orchestration
KubernetesPodOperator (container)	Low–High (depends on wrapper)	Run or model (with Cosmos)	Good (if paired with graph)	Reproducible, scalable

Quick guidance: need each dbt model as its own Airflow task with retries and downstream joins? Use Cosmos. On dbt Cloud and want Airflow to schedule/monitor but not execute? Use the provider’s DbtCloudRunJobOperator. Need a single dbt build inside an existing DAG? Bash or a container is enough. If “works on my machine” burned you last quarter, containerize.

BashOperator: direct dbt Core CLI with selectors

This is the fastest way to use Airflow with dbt. It runs dbt Core via CLI in a worker (or DockerOperator). You trade model-level observability for simplicity. You can still get partial runs with selectors and push artifacts to storage for later analysis.

# airflow dag snippet
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="dbt_build_daily",
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    dbt_build = BashOperator(
        task_id="dbt_build",
        bash_command=(
            "cd /opt/airflow/repo && "
            "dbt deps && "
            "dbt build --profiles-dir . "
            "--target prod --select tag:daily "
            "--vars '{run_date: {{ ds }}}' "
        ),
        env={
            "DBT_PROFILES_DIR": "/opt/airflow/repo",
        },
    )

To retain some visibility, persist manifest.json and run_results.json to S3/GCS keyed by {{ ds_nodash }}. Then parse them in a follow-up Python task for alerting or to annotate logs.

dbt build --select state:modified+ --defer --state s3://my-bucket/dbt_state/prod/{{ ds_nodash }}/

Failure modes you will hit: long dbt run logs timing out UI rendering, inconsistent dbt versions across workers, and a worker missing profiles. Lock a requirements.txt, pin a dbt adapter (e.g., dbt-snowflake==1.x.y), and store a profiles.yml in the container or Airflow connection. If your dbt project is large, use selectors to scope the run. For a visible per-model picture, Bash alone won’t cut it—see Cosmos below. For repo health, a dbt repo performance audit pays off quickly.

dbt Cloud API from Airflow: jobs, sensors, and artifacts

If you’re standardized on dbt Cloud, let Airflow schedule and observe a dbt job while Cloud executes. Use the provider’s operator/sensor pair. You keep dbt Cloud’s IDE, jobs, and permissions, and can still wire results back into your Airflow DAG.

from airflow import DAG
from datetime import datetime
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.providers.dbt.cloud.sensors.dbt import DbtCloudRunJobSensor

with DAG("cloud_job_daily", start_date=datetime(2024,1,1), schedule="@daily", catchup=False) as dag:
    trigger = DbtCloudRunJobOperator(
        task_id="trigger_dbt_job",
        job_id=12345,  # your dbt Cloud job id
        check_interval=10,
        timeout=60*60,
        # pass variables/selectors if your job template expects them
        account_id=6789,
    )

    wait = DbtCloudRunJobSensor(
        task_id="wait_for_dbt",
        run_id=trigger.output,  # XCom-provided run id
        timeout=60*60,
        poke_interval=30,
        deferrable=True,
    )

    trigger >> wait

Want model-level visibility in Airflow? dbt Cloud doesn’t expose each model as a native Airflow task, but you can fetch run_results.json from the dbt Cloud API after the run and dynamically map follow-up notifications or retries.

# pseudo: fetch run artifacts via Cloud API and push to XCom
# then dynamically map downstream work per failed model

Trade-offs: you rely on dbt Labs’ infrastructure for execution and retries, which is fine for most teams. You lose per-model retry semantics in Airflow itself. Cloud supports deferral/state and selector-based runs at the job level. If you need Airflow to orchestrate across multiple dbt projects and non-dbt tasks with a single directed acyclic graph, ensure your dbt Cloud job boundaries match those dependencies. If you’re migrating off Cloud or mixing, see our migration guide and dbt Cloud migration services.

Cosmos: generate Airflow tasks per dbt model

Cosmos (by Astronomer) renders your dbt graph as native Airflow tasks. Airflow understands the dbt dependency graph, so you get model-level visibility, retries, and SLAs without hand-coding hundreds of tasks. It’s the cleanest way to orchestrate dbt in Airflow with first-class observability.

# cosmos example (see official docs for full options)
from datetime import datetime
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig, RenderConfig

project = ProjectConfig(project_path="/usr/local/airflow/dags/dbt/jaffle_shop")
profile = ProfileConfig(
    profile_name="jaffle_shop",
    target_name="prod",
    profiles_yml_filepath="/usr/local/airflow/dags/dbt/profiles.yml",
)
exec_cfg = ExecutionConfig(dbt_executable_path="/usr/local/bin/dbt")
render = RenderConfig(select=["tag:daily"], exclude=["tag:skip"])

dbt_dag = DbtDag(
    dag_id="jaffle_shop_daily",
    project_config=project,
    profile_config=profile,
    execution_config=exec_cfg,
    render_config=render,
    schedule="0 2 * * *",
    start_date=datetime(2024,1,1),
    catchup=False,
)

Why operators choose Cosmos:

• Model-level retries with Airflow SLAs, on-call routing, and lineage in one place.
• Selector-based partial runs (--select, --exclude) baked into the rendered DAG.
• Works with local CLI, Docker, or Kubernetes-backed execution. Pairing with KubernetesPodOperator makes scale predictable.

Failure modes: manifest parsing times out for huge projects if SCM checkout is slow; ensure your git repository is available before parse. Keep dbt/adapter versions pinned. If you refactor models or rename a dbt model, expect task id changes—coordinate with Alerting. Start with a subset (tag:daily) before you render the whole graph. For docs and examples, see Cosmos on GitHub and our Airflow orchestration articles.

Container execution: KubernetesPodOperator and friends

Running dbt in a container gives reproducibility across workers. Whether you orchestrate with Bash, Cosmos, or Cloud, containers keep “works on my laptop” out of prod. On Kubernetes, use KubernetesPodOperator. On ECS, use an equivalent. Bake dbt, your adapter, and profiles.yml into the image or mount them. Use a read-only git checkout at tag/sha for deterministic runs.

# Airflow + KubernetesPodOperator example
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

run = KubernetesPodOperator(
    task_id="dbt_build_pod",
    name="dbt-build",
    image="ghcr.io/acme/dbt-snowflake:1.7.0",
    cmds=["bash", "-lc"],
    arguments=[
        "git clone --depth 1 --branch $GIT_REF $GIT_URL repo && "
        "cd repo && dbt deps && "
        "dbt build --target prod --select tag:daily "
        "--state s3://bucket/dbt_state/prod/{{ ds_nodash }}/ "
        "--defer"
    ],
    env_vars={
        "GIT_URL": "https://github.com/acme/analytics-repo.git",
        "GIT_REF": "refs/tags/prod-2026-08-06",
    },
    get_logs=True,
    is_delete_operator_pod=True,
)

Tips that survive production:

• Pin dbt and adapter versions in the image; do not pip install dbt at runtime.
• Put warehouse creds in Kubernetes secrets; map DBT_PROFILES_DIR to a mounted volume.
• Artifact/state goes to object storage; large logs go to stdout for Airflow to stream.
• If you orchestrate multiple repositories, use image-per-repo to decouple dependencies.

If you need per-model visibility while keeping container isolation, pair this with Cosmos’ Kubernetes execution. For managed control planes, weigh Managed Airflow options like Astronomer vs MWAA vs Composer; cost and cluster limits change your deployment shape more than code does.

Selectors and deferral: partial runs that actually work

Teams want to run only what changed. That’s dbt’s state and deferral features: compile against a prior manifest and select changed nodes plus dependencies. This works with dbt Cloud and dbt Core.

# typical command when you run dbt
# assumes prior artifacts exist at s3://my-bucket/dbt_state/prod/20260805/
dbt build --select state:modified+ --defer \
  --state s3://my-bucket/dbt_state/prod/{{ macros.ds_add(ds, -1) }}/

Airflow wiring pattern:

1. After each successful build, upload manifest.json and run_results.json to a dated prefix.
2. At the next run, compute “yesterday’s” prefix and point --state there.
3. Use selectors (tag:, +, @) to include parents/children as needed; see dbt selection syntax.

# pseudo Airflow TaskFlow to compute state path
from airflow.decorators import task

@task
def state_path(execution_date: str) -> str:
    # compute prior execution date path; handle first-run bootstrap
    return f"s3://my-bucket/dbt_state/prod/{execution_date}"

Failure modes: forgetting to persist artifacts on success only (deferral against a failed manifest causes noisy diffs), mixing targets between dev/prod but sharing a state location, and selectors that over-include causing a near-full dbt run anyway. Measure the changed-node count on your system by comparing manifest.json hashes across runs. If incremental models behave oddly after schema changes, prefer dbt build --full-refresh --select state:modified+ on that branch; see our incremental models guide.

Surface per-model failures in the Airflow UI

If you can’t adopt Cosmos yet, you can still lift model failures up into Airflow using dynamic task mapping after a single run-task. The pattern: run dbt, upload run_results.json, parse it, then fan out a lightweight task per failed model (alerts, tickets, retries via a follow-up selector).

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

@dag(start_date=datetime(2024,1,1), schedule="@daily", catchup=False, dag_id="dbt_with_failures")
def pipeline():
    @task
    def run_dbt() -> str:
        # shell out or call a Kubernetes task; return S3 key of run_results.json
        # ... run dbt build ...
        return f"s3://bucket/dbt_state/prod/{{{{ ds_nodash }}}}/run_results.json"

    @task
    def parse_failures(s3_key: str) -> list[str]:
        s3 = boto3.client("s3")
        bucket, key = s3_key.replace("s3://", "").split("/", 1)
        body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
        rr = json.loads(body)
        return [r["unique_id"] for r in rr.get("results", []) if r.get("status") != "success"]

    @task
    def notify(model_id: str):
        print(f"dbt model failed: {model_id}")

    key = run_dbt()
    failed = parse_failures(key)
    notify.expand(model_id=failed)

pipeline()

This doesn’t give you true per-model retries of the original data transformation, but it makes failures obvious and actionable. For retries, add a mapped task that runs dbt build --select {{ model_id }} against the same state and target, or record tickets with model metadata. If the team is on call for production Airflow DAGs with observability, you want this fan-out on day one. For deeper tuning, see why your dbt run is slow.

Recommendations by team situation (with code and caveats)

Small team, single warehouse (e.g., Snowflake), mostly scheduled dbt builds: Use BashOperator or DockerOperator, persist artifacts, add the failure fan-out. Code sample above is enough. Add --select tag:daily and a weekly full rebuild via a toggle env var.

On dbt Cloud and want Airflow to coordinate upstream/downstream tasks: Use DbtCloudRunJobOperator + Sensor. After success, fetch artifacts and run dbt docs generate or downstream tests. Enable deferral in the dbt Cloud job; use job params for scoped runs.

Need model-level visibility, SLAs, and dependency joins with upstream ingestion: Add Cosmos. Render dags from your dbt graph, then connect Airflow datasets or ExternalTaskSensor to upstream loaders. Example: trigger a subset with select=['source:raw.customers+'] after ingestion completes.

Have to scale across repos and environments with strict reproducibility: Use KubernetesPodOperator. Build an image per repo, include profiles.yml and a profiles_dir path, and run dbt build with state from object storage. If you also want model-level retries, wrap with Cosmos’ Kubernetes execution. For orgs deciding on control planes, read Managed Airflow compared.

Training and enablement: align on best practices for selectors, tests, and CI. If you need help to configure and deploy a robust Airflow DAG, we build them with on-call in mind: Airflow DAGs services.

FAQ: Airflow and dbt in production

What is Airflow and dbt? Is dbt an ETL tool?

Airflow schedules and orchestrates workflows; dbt builds SQL-based transformations. dbt is not an extractor or loader; it focuses on transforms, tests, and documentation. You still need ingestion. That’s why you orchestrate them together in a single workflow or data pipeline.

What’s the difference between dbt Core and Airflow?

dbt Core compiles Jinja+SQL into warehouse SQL and executes via adapters. Airflow is the orchestrator that runs tasks and manages dependencies. You can use dbt Core with Airflow (a.k.a. dbt core with airflow) for full control.

How do I connect dbt to Airflow?

Pick one: run dbt commands in a task (CLI), trigger a dbt job via Cloud operator, or render per-model tasks with Cosmos. Configure warehouse creds in profiles.yml, store the dbt project in your git repository, and set up connections. See also Airflow best practices.

How can a complex dbt DAG be displayed in Airflow?

Use Cosmos to render the dbt dependency graph into an Airflow DAG. Airflow then shows each node with native dependencies; you get retries and SLAs at the model level.

Automate full refreshes after schema changes?

Detect schema drift (e.g., compare information_schema to expected columns) and conditionally run dbt build --full-refresh --select state:modified+ for affected models. Keep this scoped; don’t full-refresh your 40M-row orders table on an X-Small warehouse unless you must. Measure by model run time and bytes scanned on your own system.

Where to find examples and best practices?

Start with the dbt articles and Airflow orchestration articles we’ve field-tested. The dbt community and docs are excellent; prefer patterns proven in production over clever macros.

Deferred runs in dbt Cloud?

Enable “Use deferral” on the job, point to a reference job for state, and scope with selectors. In Airflow, pass runtime variables if you templatize job params.

Can I use dbt Core and get Cloud-like functionality?

Mostly, with extra plumbing: containers, artifact storage, Cosmos for visibility, and CI for PR builds. You won’t get the Cloud IDE, but you gain control in open-source tooling.

Right use case for the incremental strategy?

Large fact-like tables with stable keys and append-only or upsert semantics. Test rigorously, include unique_key and filters, and have a revert path. See our incremental guide.

How do we connect dependent DAGs?

Use Airflow Datasets or ExternalTaskSensor to connect upstream ingestion to a dbt DAG. With Cosmos, you can also select only models downstream of the changed source.

How does dbt run Jinja against my database?

dbt compiles Jinja-templated SQL into plain SQL in target/, using your adapter (e.g., Snowflake). Airflow just orchestrates; the adapter executes queries. That’s how you use dbt cleanly inside orchestration.

---

Need an operator who’s been paged for this? We build and tune Airflow DAGs, dbt repos for speed, and migrations off brittle setups. Start a project.