Using dbt Artifacts in Orchestration Tools Like Airflow and Prefect

Running dbt build as a single opaque step inside an Airflow task works, but it throws away almost everything covered in dbt Artifacts — the actual per-model results, timing, and dependency structure that dbt already computed. Orchestration tools that read these artifacts directly can make much smarter decisions than “did the shell command exit zero or not.”

The Problem With Treating dbt as a Black Box

A typical naive Airflow integration:

# A common but limited pattern
BashOperator(
    task_id="run_dbt",
    bash_command="dbt build"
)

This tells Airflow exactly one thing: did the whole dbt build invocation succeed or fail. If one model out of two hundred fails, the entire task is marked failed, with no visibility into which model, why, or whether the failure was a real data problem versus a transient warehouse connection blip — all of that detail exists in run_results.json, but this pattern never reads it.

Reading run_results.json for Granular Status

import json

def parse_dbt_results(run_results_path="target/run_results.json"):
    with open(run_results_path) as f:
        results = json.load(f)

    failures = [
        r for r in results["results"]
        if r["status"] in ("fail", "error")
    ]
    return failures

failures = parse_dbt_results()
for f in failures:
    print(f"{f['unique_id']}: {f['status']} ({f['message']})")

This turns a single pass/fail signal into an actionable list — which specific models or tests failed, and their error messages — that can be surfaced directly in an alerting channel instead of just “the dbt task failed, check the logs.”

Per-Model Task Granularity With dbt-airflow Integrations

Rather than one giant dbt build task, tools purpose-built for this (Astronomer Cosmos, dbt-airflow, or manually parsing manifest.json to generate individual Airflow tasks) turn each dbt model into its own task node in the DAG, mirroring dbt’s own dependency graph as native Airflow task dependencies.

# Conceptual pattern: manifest.json informs individual task creation
import json

with open("target/manifest.json") as f:
    manifest = json.load(f)

for node_id, node in manifest["nodes"].items():
    if node["resource_type"] == "model":
        # Create an Airflow task for this specific model,
        # with upstream/downstream wired from node["depends_on"]
        pass

This gives you per-model retry logic, per-model alerting, and a DAG visualization in Airflow that mirrors dbt’s actual dependency structure — a single failed model shows as a single failed task node, not an opaque failure of the entire pipeline.

Selective Runs Based on Changed Files

A common CI/CD pattern (also covered in dbt CI/CD Integration) uses manifest.json to determine which models were actually affected by a given code change, then scopes the run accordingly rather than rebuilding the entire project on every pull request.

# Using dbt's built-in state comparison against a previous manifest
dbt build --select state:modified+ --state ./previous_manifest

This compares the current project against a saved prior manifest.json and runs only models that changed plus everything downstream of them — a meaningfully different (and much faster) execution than a full rebuild on every commit.

Feeding run_results.json Into Monitoring Dashboards

Beyond alerting on failure, teams commonly ingest run_results.json timing data into a monitoring system to track model performance over time — the same comparison approach outlined in dbt Artifacts, applied continuously rather than as a one-off diff.

import json
import time

with open("target/run_results.json") as f:
    results = json.load(f)

for r in results["results"]:
    # Push to your metrics system of choice
    emit_metric(
        metric_name="dbt_model_execution_time",
        value=r["execution_time"],
        tags={"model": r["unique_id"], "status": r["status"]}
    )

Over time, this surfaces genuinely useful trends — a model whose execution time has crept up 400% over two months is a much stronger signal in a time-series dashboard than in a single run’s console output.

Sources.json for Freshness-Gated Scheduling

An orchestration job can check sources.json (produced by dbt source freshness, covered in Source Freshness) before deciding whether to proceed with the rest of the pipeline at all.

import json

with open("target/sources.json") as f:
    freshness = json.load(f)

stale_sources = [
    r for r in freshness["results"] if r["status"] == "error"
]

if stale_sources:
    raise Exception(f"Stale sources detected: {stale_sources}")

This turns freshness from a check that just produces log output into an actual gate that can halt an orchestrated pipeline before wasting compute on transformations built on top of known-stale data.

Combining Multiple Artifacts for Richer Alerts

The most useful orchestration integrations don’t rely on a single artifact in isolation — combining run_results.json’s execution status with manifest.json’s ownership metadata (if you’ve tagged models by team, as covered in dbt Tags) lets an alert route directly to the right team rather than a generic data-engineering channel.

import json

with open("target/manifest.json") as f:
    manifest = json.load(f)
with open("target/run_results.json") as f:
    results = json.load(f)

for r in results["results"]:
    if r["status"] in ("fail", "error"):
        node = manifest["nodes"].get(r["unique_id"], {})
        tags = node.get("tags", [])
        team_tag = next((t for t in tags if t.startswith("team_")), "data-eng")
        route_alert(channel=team_tag, message=f"{r['unique_id']} failed: {r['message']}")

This turns a single generic pipeline-failure notification into a routed, team-specific alert — a meaningfully faster path to resolution than a shared channel everyone has to triage manually.

Common Mistakes

Treating dbt build as an opaque black box in orchestration. This discards almost all the granular, actionable information dbt already computed and wrote to target/ — parsing artifacts directly unlocks per-model retries, targeted alerts, and selective runs.

Not versioning or schema-checking artifacts consumed by custom tooling. As noted in dbt Artifacts, the JSON schema changes across dbt versions — orchestration code parsing these files needs to account for that, not assume a fixed structure forever.

Rebuilding the entire project on every scheduled run when state-based selection would be faster and cheaper. For large projects, --select state:modified+ against a saved manifest is often a significant, low-effort performance win.

Summary

Reading dbt’s artifacts directly instead of treating the whole invocation as pass/fail is what turns a basic scheduled job into an orchestration setup with real per-model observability, selective execution, and meaningful alerting.