Building a dbt and Dagster Pipeline That Treats Data as a First-Class Asset, Not a Task Graph

TL;DR: The common way to orchestrate dbt is to wrap dbt run in a single task and schedule it on a cron. That throws away the dependency graph dbt already knows. A better architecture maps every dbt model to a Dagster software-defined asset, turns dbt tests into asset checks, drives execution with declarative automation conditions instead of fixed schedules, and uses partitions for incremental models. You get per-table lineage, per-table retries, and freshness guarantees. The cost is some upfront modelling discipline and a few sharp edges covered below.

The problem with "dbt as one big task"

Most dbt deployments start the same way: a cron job or an Airflow BashOperator that runs dbt build against the whole project once an hour or once a night. It works until it doesn't, and the ways it fails are predictable:

• All-or-nothing granularity. One failing model in a 400-model project marks the entire run red. Re-running means re-running everything, or someone hand-crafting a --select string at 2am.
• Invisible upstream coupling. The ingestion job that loads raw.orders and the dbt job that reads it are two separate schedules that happen to line up. When ingestion runs late, dbt happily rebuilds on stale data and nobody notices until a dashboard is wrong.
• Two lineage graphs. dbt knows model-to-model lineage. The orchestrator knows task-to-task lineage. Neither knows the full path from a source API to a BI extract.
• Tests without consequences. dbt test failures show up in logs, but nothing structurally stops a downstream Python ML feature job from consuming a table whose uniqueness test just failed.

The root issue is that the unit of orchestration is an action ("run dbt") rather than the thing you care about ("the fct_orders table, fresh as of 06:00, passing its tests"). Dagster's software-defined assets flip that: you declare the tables, files, and models that should exist, their upstream dependencies, and the code that produces them. The orchestrator's job becomes keeping those assets in their desired state. We compared the two philosophies at the platform level in our Airflow vs. Dagster decision framework; this article is about the implementation once you have chosen the asset model.

Architecture overview

The target architecture has five layers, all expressed in one asset graph:

1. Ingestion assets — Python assets (or integration-generated assets from tools like Fivetran, Airbyte, or dlt) that land raw tables. Their asset keys must match the keys Dagster derives for the corresponding dbt source entries.
2. dbt assets — generated from manifest.json by the @dbt_assets decorator: one Dagster asset per dbt model, seed, and snapshot, with dependencies taken from ref() and source() calls.
3. Asset checks — dbt tests on models (and optionally sources) loaded as Dagster asset checks, plus any custom Python checks that dbt can't express well.
4. Downstream non-dbt assets — ML features, reverse-ETL syncs, BI extracts, declared with explicit dependencies on dbt asset keys.
5. Automation and freshness — automation conditions that decide when each asset should materialize, and freshness policies that say how stale is too stale.

The key property: there is exactly one graph. A change to stg_orders shows its blast radius all the way from the source API to the reverse-ETL sync that pushes customer segments into your CRM.

Layer 1: Loading the dbt project as assets

The core integration is small. DbtProject points at the dbt project and manages manifest.json; @dbt_assets reads the manifest and emits one asset per node; DbtCliResource runs the CLI and streams structured events back to Dagster.

from pathlib import Path
from dagster import AssetExecutionContext, Definitions
from dagster_dbt import DbtCliResource, DbtProject, dbt_assets

analytics_project = DbtProject(project_dir=Path(__file__).parent / "analytics")
analytics_project.prepare_if_dev()  # parses the manifest locally; CI builds it in prod

@dbt_assets(manifest=analytics_project.manifest_path)
def analytics_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

defs = Definitions(
    assets=[analytics_dbt_assets],
    resources={"dbt": DbtCliResource(project_dir=analytics_project)},
)

Two details matter more than they look:

The manifest is a build artifact, not a runtime computation. prepare_if_dev() is a convenience for local development. In production, generate manifest.json in CI (dbt parse or dbt compile) and ship it with the code location image. If the manifest is parsed at import time in production, code-location load time grows with project size and a broken profiles.yml breaks the whole deployment, not one run.

dbt build is scoped per run. Even though the function body says ["build"], the context passed to dbt.cli tells the resource which subset of assets Dagster actually selected, and the integration translates that into a dbt selection. Materialize one model from the UI and only that model (and its tests) runs. This is what gives you per-table retries for free.

Layer 2: The translator is your contract

DagsterDbtTranslator controls how dbt node properties turn into Dagster asset properties: asset key, group, owners, tags, metadata, description, and automation condition. Treat it as the contract between the dbt project and the rest of the platform, and decide its rules deliberately rather than accepting defaults and patching later.

from typing import Any, Mapping, Optional
from dagster import AssetKey, AutomationCondition
from dagster_dbt import DagsterDbtTranslator, DagsterDbtTranslatorSettings

class PlatformTranslator(DagsterDbtTranslator):
    def get_asset_key(self, props: Mapping[str, Any]) -> AssetKey:
        # Namespace by warehouse so keys stay unique if a second dbt project appears
        return super().get_asset_key(props).with_prefix("warehouse")

    def get_group_name(self, props: Mapping[str, Any]) -> Optional[str]:
        # Use the dbt folder layer (staging / intermediate / marts) as the group
        fqn = props.get("fqn", [])
        return fqn[1] if len(fqn) > 2 else "default"

    def get_automation_condition(self, props: Mapping[str, Any]) -> Optional[AutomationCondition]:
        if "realtime" in props.get("tags", []):
            return AutomationCondition.eager()
        return AutomationCondition.on_cron("0 * * * *")

translator = PlatformTranslator(
    settings=DagsterDbtTranslatorSettings(
        enable_source_tests_as_checks=True,
        enable_code_references=True,
    )
)

Rules we would set early:

Asset keys can also be overridden per node using meta.dagster.asset_key in dbt YAML, which is useful for the handful of sources that don't fit your naming convention.

A caveat on the last row: at the time of writing, Dagster does not read automation conditions directly from dbt YAML config (there is an open request for it), so the translator pattern above, keyed on dbt tags or meta, is the practical bridge.

Layer 3: Connecting ingestion to dbt sources

The graph only becomes end-to-end when the raw tables are assets too. For a source table loaded by your own code, declare an asset whose key is derived from the dbt source, rather than typing the key by hand:

from dagster import asset
from dagster_dbt import get_asset_key_for_source

@asset(key=get_asset_key_for_source([analytics_dbt_assets], "shop"))
def shop_orders_raw(context) -> None:
    # Pull from the orders service's ingestion APIs and load into raw.shop_orders
    ...

get_asset_key_for_source handles a source with a single table; for a source with multiple tables, get_asset_keys_by_output_name_for_source lets one multi-asset produce all of them. Either way, deriving the key from the dbt definitions means a rename in dbt surfaces as a load-time error rather than a silently orphaned asset.

Once ingestion and dbt share keys, "run dbt after ingestion" stops being a scheduling coincidence and becomes a dependency the orchestrator can reason about.

Layer 4: dbt tests as asset checks

Dagster loads dbt tests on models as asset checks automatically. Generic tests (unique, not_null, relationships, accepted_values) attach to the model they test. Singular tests that reference several models need a meta.dagster.ref config to say which asset they belong to; without it, Dagster still runs the test but records the result as an observation rather than a check.

Why this matters architecturally:

• Checks are first-class results with severity. A dbt test configured with severity: error fails the check; severity: warn records a warning. Downstream automation can then be written to skip materialization when an upstream asset has a failing blocking check, instead of propagating bad data.
• Source tests become upstream gates. With enable_source_tests_as_checks=True, not_null and freshness-style tests on sources attach to the ingestion assets. A broken load is visible on the raw asset, not discovered three layers downstream.
• Selection is precise. Dagster uses dbt's indirect selection mode so you can materialize assets without their checks, or re-run checks without rebuilding the model, which is useful when a check was flaky or thresholds were changed.

dbt tests are good at row-level assertions. They are weaker at statistical assertions: volume anomalies, distribution drift, "this partition has 40% fewer rows than the trailing average." Those are better written as Python asset checks alongside the dbt ones. We covered the detection side in depth in our data observability guide; in this architecture they simply become more checks on the same assets.

Tests at the dbt layer are also the enforcement point for data contracts on staging models: dbt model contracts (contract: {enforced: true}) pin column names and types, and checks on the source assets catch producer-side drift before it reaches marts.

Layer 5: Declarative automation instead of cron

This is the part that actually changes operations. Instead of "run the whole project hourly," each asset carries an AutomationCondition describing when it should be materialized:

• AutomationCondition.eager() — materialize whenever any upstream dependency updates (subject to not being in progress and not having missing or failed parents). Good for low-latency, cheap models. For time-partitioned assets, it only targets the latest partition.
• AutomationCondition.on_cron("0 6 * * *") — after each cron tick, materialize once all upstream dependencies have updated since that tick. Good for daily marts that should reflect a complete day of every input, not whichever source arrived first.
• Custom compositions — conditions are composable with &, |, and ~, and built from primitives such as "newly missing," "any deps updated," "in progress," and "any deps in progress." Dagster's docs describe customizing eager and on_cron rather than writing conditions from scratch, which is the sensible starting point.

The difference from cron is subtle but important. on_cron doesn't mean "run at 06:00"; it means "after 06:00, run once you have everything you need." If ingestion is late, dbt waits instead of building on yesterday's data. If ingestion never arrives, nothing runs and the freshness policy (below) raises the alarm, which is the right failure mode.

A practical tiering scheme:

Because the tier lives in dbt as a tag, analytics engineers change cadence in the same pull request as the model logic.

Freshness as an SLA, not a schedule

Automation says when to try. Freshness says when to worry. Since Dagster 1.12, freshness policies replace the older freshness checks; FreshnessPolicy.time_window(fail_window=..., warn_window=...) marks an asset WARN or FAIL based on time since last materialization, and a cron-based variant exists for "must be updated by 07:00." Attach policies to the marts that humans read, not every staging view, or alert fatigue sets in within a week.

Incremental models and partitions

For large fact tables, rebuilding everything is wasteful and dbt incremental models with is_incremental() are the standard answer. The weakness of plain incremental models is that "what's new" is computed from the target table's own max timestamp. Late-arriving data, backfills, and reprocessing a specific bad day are all awkward.

Dagster partitions make the time window explicit. Pass a PartitionsDefinition to @dbt_assets, read the partition's time window from the context, and hand it to dbt as vars:

import json
from dagster import AssetExecutionContext, DailyPartitionsDefinition
from dagster_dbt import DbtCliResource, dbt_assets

daily = DailyPartitionsDefinition(start_date="2024-01-01")

@dbt_assets(
    manifest=analytics_project.manifest_path,
    select="tag:partitioned",
    partitions_def=daily,
    dagster_dbt_translator=translator,
)
def partitioned_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    window = context.partition_time_window
    dbt_vars = {"min_date": window.start.isoformat(), "max_date": window.end.isoformat()}
    yield from dbt.cli(["build", "--vars", json.dumps(dbt_vars)], context=context).stream()

@dbt_assets(
    manifest=analytics_project.manifest_path,
    exclude="tag:partitioned",
    dagster_dbt_translator=translator,
)
def unpartitioned_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

And in the model:

{{ config(materialized='incremental', unique_key='order_id', tags=['partitioned']) }}

select * from {{ ref('stg_orders') }}
{% if is_incremental() %}
  where order_ts >= '{{ var("min_date") }}' and order_ts < '{{ var("max_date") }}'
{% endif %}

Notes from doing this in practice:

• All assets in one @dbt_assets call share one PartitionsDefinition. That is why the example splits the project in two with select/exclude on a tag. Keep the tag the single source of truth.
• Make the partitioned models idempotent per window. Use unique_key with a merge or a delete+insert strategy so re-running a partition replaces its rows instead of duplicating them. This is what makes backfills safe.
• Backfills become first-class. Reprocessing March 3–9 is a partition-range backfill in the UI rather than a manually edited vars string.
• Partition-aware checks. Asset checks can be partitioned too, so a failing volume check points at the specific day that is wrong.

Downstream non-dbt assets

Anything that consumes dbt output should depend on the dbt asset key explicitly:

from dagster import asset, AutomationCondition
from dagster_dbt import get_asset_key_for_model

@asset(
    deps=[get_asset_key_for_model([unpartitioned_dbt_assets], "fct_customer_ltv")],
    automation_condition=AutomationCondition.on_cron("30 6 * * *"),
)
def churn_features() -> None:
    ...

Now the ML feature job waits for the mart, inherits its lineage, and can be blocked by its failing checks. The same pattern applies to reverse-ETL syncs and BI extract refreshes.

Failure modes and how to design against them

1. Manifest drift between deploy and runtime. If the manifest shipped with the code location doesn't match the dbt project on disk, Dagster's selection and dbt's view of the graph disagree. Build the manifest in CI from the same commit, and fail the deploy if dbt parse fails.

2. Automation stalls after code reloads. There have been reported issues of eager() conditions stopping after a code-location reload triggered by manifest changes. Treat automation as something to monitor: alert on freshness, not just on failures, so a silent stall is caught by the SLA layer.

3. Asset key collisions and renames. Two dbt projects with a stg_orders model, or a model rename, break key identity. Prefix keys per project in the translator and treat key changes as migrations.

4. Over-eager fan-out. Putting eager() on everything means a frequently updated source can trigger hundreds of downstream rebuilds per hour, which is expensive on consumption-priced warehouses. Default to on_cron, and reserve eager for small models where latency actually matters.

5. Incremental logic that isn't idempotent. A partitioned run that appends rather than merges will double-count on retry. Test this explicitly: run the same partition twice in CI against a scratch schema and assert row counts are stable.

6. Tests that never block anything. If every dbt test is severity: warn, checks become noise. Decide which tests are blocking (primary keys, referential integrity on marts) and which are informational, and document it.

7. Monolithic dbt invocations. A single @dbt_assets over a very large project still works, but runs can be slow to start as dbt parses and compiles. Splitting by domain into separate @dbt_assets definitions (or separate code locations) keeps selection fast and isolates failures.

Adding observability without extra tooling

The integration can emit more than pass/fail. Calling .fetch_row_counts() on the event stream records row counts as materialization metadata, fetched in parallel with the run, and .fetch_column_metadata() records column schemas and column-level lineage:

yield from (
    dbt.cli(["build"], context=context)
    .stream()
    .fetch_row_counts()
    .fetch_column_metadata()
)

Row counts over time give you a cheap volume signal per table before you invest in anomaly detection, and column lineage makes impact analysis for a column rename a lookup rather than a grep. For CI, passing a state_path to DbtProject and adding dbt.get_defer_args() lets branch deployments build only changed models and defer the rest to production, which keeps PR checks fast.

Decision checklist

Before rolling this out, answer these explicitly:

• [ ] What is the asset key scheme, and does it match ingestion asset keys for every dbt source?
• [ ] Where is manifest.json built, and does the deploy fail if parsing fails?
• [ ] Which dbt tags map to which automation conditions? Is the default on_cron?
• [ ] Which marts have freshness policies, and who is paged?
• [ ] Which dbt tests are blocking (error) and which are informational (warn)?
• [ ] Which models are partitioned, and are they idempotent per partition?
• [ ] Are downstream Python/ML/reverse-ETL assets declared with explicit deps on dbt keys?
• [ ] Is the project split so a failure in one domain doesn't block others?

Sources

• Dagster docs: dagster-dbt integration reference
• Dagster docs: dagster-dbt API reference (DagsterDbtTranslator, settings)
• Dagster docs: Declarative Automation and Automation condition reference
• Dagster docs: Adjust dbt asset config for incremental models
• Dagster docs: Testing assets with asset checks
• Dagster docs: Asset freshness policies
• Dagster blog: What Are Software-Defined Assets?
• GitHub: dagster-io/dagster issue #25866 — automationcondition in dbt metadata; issue #33656 — eager() after code location reload
• dbt docs: Indirect selection, Defer

Syslabs' engineering team designs and builds this kind of custom data platform for clients moving from scheduled scripts to asset-based orchestration.