dagster-dbt

dagster-dbt reads dbt 37,942 's manifest.json and turns every model, seed and snapshot into an asset and every dbt test into an asset check, so Analytical SQL and Data Warehouses's project joins the asset graph unchanged. DbtProject locates the project (prepare_if_dev() runs dbt parse under dagster dev; production builds the manifest in CI), and @dbt_assets defines what one dbt command materializes. fct_sales is built per day, so it gets its own:

definitions.py (excerpt): dbt models as assets, and a blocking checkPython

class BookNestTranslator(DagsterDbtTranslator):    # fct_sales must wait for the day's load
    def get_asset_spec(self, manifest, unique_id, project):
        spec = super().get_asset_spec(manifest, unique_id, project)
        if spec.key == dg.AssetKey("fct_sales"):
            spec = spec.merge_attributes(deps=[["shop", "orders"], ["shop", "order_items"]])
        return spec
@dbt_assets(manifest=dbt_project.manifest_path, select="fct_sales", partitions_def=daily,
            backfill_policy=dg.BackfillPolicy.multi_run(1),       # one day per run
            dagster_dbt_translator=BookNestTranslator())
def sales_facts(context: dg.AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()   # dbt build --select fct_sales
@dg.asset_check(asset="fct_sales", partitions_def=daily, blocking=True)
def fct_sales_matches_source(context: dg.AssetCheckExecutionContext, db: BookNestDB):
    try:
        lines, gross = p.check_day(context.partition_key, db.connect())
    except ValueError as e:                      # a failed check blocks daily_genre_sales
        return dg.AssetCheckResult(passed=False, metadata={"error": str(e)})
    return dg.AssetCheckResult(passed=True, metadata={"lines": lines, "gross": float(gross)})

A second @dbt_assets with exclude="fct_sales" covers the staging views, dimensions, seed and snapshot. dbt's sources map to asset keys [source, table], so source('shop', 'orders') is the shop/orders asset of Software-Defined Assets and the lineage joins up by itself. The code location ended up with 11 checks: ten dbt tests (such as fct_sales_book_fk) and fct_sales_matches_source.

The first real run explains the translator. The job selects only fct_sales from the dbt project; its upstream path runs through stg_orders and stg_order_items, which the job does not include, and Dagster 177,056 orders steps only along edges inside the selection. dbt therefore ran before the load, found nothing new, and the blocking check stopped the run before publish:

Output of 36

STEP_SUCCESS             sales_facts       Finished execution of step "sales_facts" in 13.39s.
ASSET_MATERIALIZATION    shop_tables       Materialized value shop orders.
ASSET_CHECK_EVALUATION   fct_sales_fct_... Asset check 'fct_sales_matches_source' ...
{'error': TextMetadataValue(text='2026-06-28: fact 0/0 vs source 269/6855.25')}

Partitioned asset checks are a preview feature in 1.13 (Dagster warns), but blocking worked like check in the Airflow 129 DAG.

MENU