01 - dbt: Airflow Integration

11 min read

dbt: Airflow Integration

Quote

“It becomes even more important to have something like Airflow that brings everything together in a sane place where every little piece of the puzzle can be orchestrated properly.”

Source: Maxime Beauchemin | creator of Apache Airflow

Airflow BashOperator Wrapping dbt run

The simplest approach: invoke the dbt CLI as a shell command from within an Airflow task. The entire dbt project runs as a single Airflow task.

from airflow.operators.bash import BashOperator
 
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command=(
        "cd /opt/dbt/financial_indices && "
        "dbt run --target prod --vars '{run_date: {{ ds }}}'"
    ),
    env={
        "DBT_BIGQUERY_PROJECT": "fin-data-prod",
        "DBT_BIGQUERY_DATASET": "esg_transformed",
        "GOOGLE_APPLICATION_CREDENTIALS": "/secrets/sa-key.json",
    },
)

Pros

  • Zero extra dependencies beyond the dbt CLI on the worker.
  • Fast to implement; works with any dbt adapter.

Cons

  • One monolithic task: a single model failure fails the entire block.
  • No per-model retry, duration metrics, or partial re-run from Airflow.
  • Log output is a single stream; hard to isolate failures.

When to use the single-task approach

Use it for non-critical pipelines or when the dbt project is small enough that one coarse retry surface is operationally acceptable.

Option 2: astronomer-cosmos (Each Model = Airflow Task)

astronomer-cosmos parses the dbt project’s manifest.json at DAG parse time and generates one Airflow task per dbt node (model, seed, snapshot, test). Dependencies between tasks mirror the dbt DAG.

Airflow astronomer-cosmos Installation

pip install astronomer-cosmos[dbt-bigquery]

Airflow astronomer-cosmos DAG Definition

from datetime import datetime
from datetime import timedelta
from pathlib import Path
 
from airflow import DAG
from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import GoogleCloudOauthProfileMapping
 
profile_config = ProfileConfig(
    profile_name="financial_indices",
    target_name="prod",
    profile_mapping=GoogleCloudOauthProfileMapping(
        conn_id="google_cloud_default",
        profile_args={
            "project": "fin-data-prod",
            "dataset": "esg_transformed",
            "location": "EU",
        },
    ),
)
 
esg_dbt_dag = DbtDag(
    dag_id="esg_dbt_cosmos",
    project_config=ProjectConfig(Path("/opt/dbt/financial_indices")),
    profile_config=profile_config,
    execution_config=ExecutionConfig(dbt_executable_path="/usr/local/bin/dbt"),
    operator_args={
        "vars": {"run_date": "{{ ds }}"},
        "retries": 2,
        "retry_delay": timedelta(seconds=30),
    },
    schedule="0 4 * * *",
    start_date=datetime(2025, 1, 1),
    catchup=False,
)

Pros

  • Full per-model task observability in the Airflow UI.
  • Retry individual failed models without re-running the whole project.
  • Test nodes appear as separate tasks immediately downstream of their model.

Cons

  • DAG parse time increases with project size (mitigate with dbt ls caching).
  • Requires the manifest to be present at parse time; coordinate with CI/CD.
  • Additional dependency (astronomer-cosmos) must be pinned and managed.

Cosmos load modes

Cosmos supports LoadMode.DBT_LS for runtime discovery and LoadMode.MANIFEST for pre-built metadata. Prefer MANIFEST in production when you want predictable DAG parse behavior and fewer moving parts on the scheduler.

Airflow CloudRunJobOperator (Isolated Container)

Run dbt inside a Cloud Run Job, treating the entire dbt invocation as a containerised ephemeral workload. Airflow submits the job and polls for completion.

from airflow.providers.google.cloud.operators.cloud_run import CloudRunExecuteJobOperator
 
dbt_cloud_run = CloudRunExecuteJobOperator(
    task_id="dbt_run_container",
    project_id="fin-data-prod",
    region="europe-west1",
    job_name="dbt-esg-transformer",
    overrides={
        "containerOverrides": [
            {
                "name": "dbt-runner",
                "args": ["run", "--target", "prod", "--vars", "{run_date: {{ ds }}}"],
                "env": [
                    {"name": "RUN_DATE", "value": "{{ ds }}"},
                ],
            }
        ]
    },
    gcp_conn_id="google_cloud_default",
)

Pros

  • Complete isolation: no dbt installation on Airflow workers.
  • Independently scalable compute; Cloud Run handles cold start automatically.
  • Container image version is pinned, enabling atomic rollbacks.

Cons

  • Cold start latency (10–30 s per invocation) adds to total pipeline duration.
  • Observability is limited to Airflow task-level (pass/fail), not per-model.
  • Cloud Run Job logs must be viewed separately in Cloud Logging.

Airflow dbt Operator Comparison Table

DimensionBashOperatorastronomer-cosmosCloudRunJobOperator
ComplexityLowMediumMedium–High
Per-model retryNoYesNo
ObservabilityStream log onlyPer-task in Airflow UIJob-level pass/fail
IsolationShared worker envShared worker envDedicated container
CostMinimalMinimal (worker CPU only)Cloud Run compute + egress
RollbackRe-run DAGRe-run specific tasksPin image version
Best forSmall projectsLarge projects, on-callRegulated/isolated envs

Full DAG: Extract to dbt Cosmos to dbt test to Publish

This pattern represents a complete ESG data pipeline: raw provider data lands in GCS, dbt transforms it, tests validate quality, and a downstream publish step refreshes the index calculation API.

from datetime import datetime, timedelta
from pathlib import Path
 
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from cosmos import DbtTaskGroup, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import GoogleCloudOauthProfileMapping
 
default_args = {
    "owner": "data-engineering",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-oncall@example.com"],
}
 
profile_config = ProfileConfig(
    profile_name="financial_indices",
    target_name="prod",
    profile_mapping=GoogleCloudOauthProfileMapping(
        conn_id="google_cloud_default",
        profile_args={
            "project": "fin-data-prod",
            "dataset": "esg_transformed",
            "location": "EU",
        },
    ),
)
 
def extract_esg_provider_data(**context):
    """Pull raw ESG scores from provider SFTP into GCS staging bucket."""
    run_date = context["ds"]
    # ... provider client logic omitted
    context["ti"].xcom_push(key="raw_gcs_uri", value=f"gs://fin-raw/esg/{run_date}/")
 
def publish_index_snapshot(**context):
    """Notify downstream API that new index data is available."""
    run_date = context["ds"]
    raw_uri = context["ti"].xcom_pull(task_ids="extract_esg", key="raw_gcs_uri")
    # ... API call or Pub/Sub publish omitted
 
with DAG(
    dag_id="esg_index_pipeline",
    default_args=default_args,
    schedule="0 5 * * 1-5",          # weekdays at 05:00 UTC
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=["esg", "dbt", "production"],
) as dag:
 
    extract = PythonOperator(
        task_id="extract_esg",
        python_callable=extract_esg_provider_data,
    )
 
    dbt_transform = DbtTaskGroup(
        group_id="dbt_transform",
        project_config=ProjectConfig(Path("/opt/dbt/financial_indices")),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/usr/local/bin/dbt"),
        operator_args={
            "vars": {"run_date": "{{ ds }}"},
            "select": "tag:esg",          # only ESG-tagged models
        },
    )
 
    dbt_test = DbtTaskGroup(
        group_id="dbt_test",
        project_config=ProjectConfig(Path("/opt/dbt/financial_indices")),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/usr/local/bin/dbt"),
        operator_args={
            "select": "tag:esg",
            "store_failures": True,
        },
    )
 
    publish = PythonOperator(
        task_id="publish_index_snapshot",
        python_callable=publish_index_snapshot,
    )
 
    extract >> dbt_transform >> dbt_test >> publish

Passing Variables to dbt

--vars flag (inline YAML)

operator_args={"vars": {"run_date": "{{ ds }}", "provider": "msci"}}

Rendered at runtime by Airflow’s Jinja engine before dbt receives the string.

Environment Variables

dbt reads env_var('KEY') calls from profiles.yml and model SQL. Pass secrets as environment variables rather than embedding them in --vars.

# profiles.yml
financial_indices:
  target: "{{ env_var('DBT_TARGET', 'dev') }}"
  outputs:
    prod:
      type: bigquery
      project: "{{ env_var('DBT_BQ_PROJECT') }}"
      dataset: "{{ env_var('DBT_BQ_DATASET') }}"
env={"DBT_BQ_PROJECT": "fin-data-prod", "DBT_BQ_DATASET": "esg_transformed"}

XComs to dbt via --vars

When an upstream task computes a value (e.g., a reference date from a vendor feed), pass it into dbt via --vars using Airflow’s template syntax:

bash_command=(
    "dbt run --vars '{cutoff_date: {{ ti.xcom_pull(task_ids=\"extract_esg\", "
    "key=\"cutoff_date\") }}}'"
)

XCom variable constraints

XCom values pulled into --vars must be strings or simple scalars. Never pass secrets through XComs; use Airflow Connections or Secret Manager instead.

Safe variable passing

Pass secrets to dbt via environment variables using Airflow’s env parameter on BashOperator, sourced from an Airflow Connection or Secret Manager backend. Use --vars only for non-sensitive runtime parameters such as run_date or provider_code.

Handling dbt Failures in Airflow

dbt Airflow Failure Callback — on_failure_callback

from airflow.models import TaskInstance
 
def on_dbt_failure(context: dict):
    ti: TaskInstance = context["task_instance"]
    dag_id = context["dag"].dag_id
    run_id = context["run_id"]
    # Post to Slack, Datadog, or PagerDuty
    send_alert(f"dbt failed in DAG {dag_id} | run {run_id} | task {ti.task_id}")
 
default_args = {"on_failure_callback": on_dbt_failure}

Partial Re-runs with Cosmos

Because each model is a separate task, you can clear and re-run only failed tasks from the Airflow UI without re-executing upstream models that succeeded. This dramatically reduces re-run cost for large ESG transformation projects with hundreds of company-level models.

Exit Code Handling

dbt exits with a non-zero code on test failures as well as runtime errors. Distinguish them by parsing run_results.json in a TriggerRule.ALL_DONE downstream task:

from airflow.utils.trigger_rule import TriggerRule
 
check_results = PythonOperator(
    task_id="check_dbt_results",
    python_callable=parse_run_results,      # reads /opt/dbt/target/run_results.json
    trigger_rule=TriggerRule.ALL_DONE,      # runs even if dbt_run failed
)

Graph View

Backlinks