P→D Pipeline to Dashboard nihalsulaiman.com

Day 33. dbt inside Dagster, and the alternatives

⏱ 4h · learn 60m · lab 145m · ship 20m · review 15m🏷 restaurantsBefore: Day 32

By the end of today you can

Load a dbt project as Dagster assets with dagster-dbt
Connect Python assets upstream and downstream of dbt models
Partition dbt runs by business date
Compare dbt with SQLMesh and plain stored procedures

1. Why this matters

In the method you saw, Python jobs ran in Dagster and SQL did the modelling. With dagster-dbt, every dbt model becomes a Dagster asset, so the whole chain (email → clean → load → dbt models → report) is one graph with one set of schedules, checks and lineage.

2. dbt models as assets

from pathlib import Path
import dagster as dg
from dagster_dbt import DbtCliResource, DbtProject, dbt_assets

dbt_project = DbtProject(project_dir=Path(__file__).parents[2] / "restaurant_dbt")
dbt_project.prepare_if_dev()                    # builds manifest.json during `dagster dev`

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

defs = dg.Definitions(assets=[raw_pos, stg_pos_parquet, restaurant_models, daily_report],
                      resources={"dbt": DbtCliResource(project_dir=dbt_project)})

dbt data_tests become Dagster asset checks automatically; a failing unique test shows red on the model's asset.

3. Wiring Python ↔ dbt

Upstream: a dbt source maps to a Dagster asset key. Name your Python asset to match (or set meta: {dagster: {asset_key: [...]}} in _sources.yml). Now raw_pos → stg_pos__order_lines shows as an edge.
Downstream: a Python asset that sends a report depends on dg.AssetKey(["fct_restaurant_daily"]) (or use get_asset_key_for_model).

4. Partitioned dbt runs

@dbt_assets(manifest=dbt_project.manifest_path, partitions_def=daily, select="tag:daily")
def daily_models(context, dbt: DbtCliResource):
    start, end = context.partition_time_window
    vars = {"min_date": start.strftime("%Y-%m-%d"), "max_date": end.strftime("%Y-%m-%d")}
    yield from dbt.cli(["build", "--vars", json.dumps(vars)], context=context).stream()

Models read {{ var('min_date') }} in their incremental filter. Now a Dagster backfill of 60 days runs dbt for exactly those days.

Maths and principle

Principle: composing two DAGs along shared nodes gives one DAG, as long as no cycle is introduced.

Equation: \(G = (V_{py} \cup V_{dbt},\ E_{py} \cup E_{dbt} \cup E_{\text{bridge}})\) where bridge edges map sources to Python assets. \(G\) stays acyclic iff no dbt model feeds a Python asset that feeds that model's sources.

Analogy: joining two circuits at their terminals. Where it breaks: a Python step that writes back into a dbt source table (reverse ETL into the same schema) closes a loop.

Try it: Python raw_pos → dbt stg → dbt fct → Python report. Add Python fix_pos reading fct and writing raw_pos. Cycle? (Yes: raw_pos → … → fct → fix_pos → raw_pos.)

Alternatives to dbt

SQLMesh (open source): SQL models with built-in virtual environments, column-level lineage, plan/apply like Terraform, and understanding of incremental time ranges. Strong on correctness and cost of dev environments.
Stored procedures + SQL Agent / Elastic Jobs: still common in Microsoft shops. Works; weaker testing and lineage. You'll meet it, so be able to read and migrate it.
Dataform (Google, BigQuery), Fabric notebooks/dataflows (Microsoft low-code).

The 5-year lens

When you inherit a pile of stored procedures, don't rewrite everything. Wrap the procs as Dagster assets first (an asset that runs exec mart.usp_load_sales @d), add checks and lineage, then migrate the riskiest or most-changed ones to dbt one at a time with a reconciliation test comparing old vs new outputs. That migration story is a classic senior interview answer.

Lab

Add dagster-dbt to the restaurant-dbt repo (or combine with hotel-dagster style layout). Load dbt models as assets.
Map dbt sources to your Python ingest/clean assets so lineage is continuous. Screenshot the full graph.
Partition the daily models; backfill 30 days from the UI.
Make a dbt test fail on purpose (inject a duplicate) and see the asset check go red and block the report asset.
Optional: port two models to SQLMesh and write 10 lines comparing the experience.

Ship

feat(orchestration): dbt models as Dagster assets with partitions and checks.