Course 6 · Capstone

Lab 05 — Orchestrate with Dagster

So far you've been running three scripts by hand, in your head, in the right order: extract from Postgres, consume the telemetry stream, build dbt. In this lab you hand that whole sequence to Dagster. You'll declare each step as an asset, wire up the dependencies, watch Dagster figure out the order for you, and put the whole pipeline on a daily schedule. This is the moment your collection of scripts becomes a platform.

Setup — do this first

You need Dagster installed in your project's virtual environment (from Course 5) and the scripts from Labs 02–04 working: ingest/extract_postgres.py populates bronze from Postgres, ingest/consume_stream.py lands telemetry, and your dbt/ project builds silver + gold. Have the Compose stack up (docker compose up -d) so Postgres and Redpanda are reachable. Run each script once by hand first to confirm they still work — Dagster only orchestrates what already runs.

The goal

Right now your pipeline lives in your memory. You know that extract_postgres.py has to run before dbt, and that the telemetry consumer has to land data before the hourly rollup can summarize it. Forget that order — or run dbt before the extract finishes — and you get stale or empty tables. That's fine for one person on one laptop. It is a disaster for anything real.

An orchestrator takes the ordering out of your head and puts it in code. Dagster's model is assets: each asset is a thing that exists in your warehouse (a table, a view, a rollup), and you describe what each asset depends on. Dagster reads those dependencies, builds a graph, and runs everything in the only correct order — every time, with logs, retries, and a UI.

By the end of this lab, these four manual steps become one dependency-aware graph:

POSTGRES STREAM dbt ROLLUP ┌───────────────┐ ┌────────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ bronze_ │ │ telemetry_ │ │ dbt_build │ │ fct_telemetry_hourly │ │ business │ │ bronze │ │ (silver+gold)│ │ (hourly rollup) │ └──────┬────────┘ └───────┬────────┘ └──────┬───────┘ └──────────┬───────────┘ │ │ ▲ ▲ │ extract from Postgres │ consume Redpanda │ │ │ │ │ │ └───────────────────────────┴──────────────────────────┘ │ both bronze layers must land BEFORE dbt runs │ │ dbt produces dim/fct tables ──────────────────────────────────────────────────────────┘ the hourly rollup reads the telemetry that dbt staged ────────────────────────────────────────────────────────────────────────────────────────────────────── Dagster reads these arrows and runs: bronze_business + telemetry_bronze → dbt_build → fct_telemetry_hourly

You declare the boxes and the arrows. Dagster does the scheduling.

Define the assets

An asset is just a Python function decorated with @asset. The function name becomes the asset's name; its return is metadata about what it produced; and its function arguments name the assets it depends on. That argument-equals-dependency trick is the whole magic — Dagster reads the signature and builds the graph.

Create dagster_app/assets.py. We'll reuse the exact scripts you already wrote, calling them as subprocesses so there's a single source of truth for each step. (You could also import and call the functions directly — both are common; subprocess keeps the lab honest about what's actually running.)

dagster_app/assets.py

import subprocess
from pathlib import Path

from dagster import asset, AssetExecutionContext, MaterializeResult, MetadataValue
import duckdb

# Project paths — adjust if your repo root differs.
REPO = Path(__file__).resolve().parents[1]
INGEST = REPO / "ingest"
DBT_DIR = REPO / "dbt"
WAREHOUSE = REPO / "warehouse" / "griddp.duckdb"


def _run(context: AssetExecutionContext, cmd: list[str], cwd: Path) -> str:
    """Run a subprocess, stream its output into Dagster's logs, raise on failure."""
    context.log.info(f"$ {' '.join(cmd)}  (cwd={cwd})")
    result = subprocess.run(
        cmd, cwd=cwd, capture_output=True, text=True
    )
    if result.stdout:
        context.log.info(result.stdout.strip())
    if result.returncode != 0:
        context.log.error(result.stderr.strip())
        raise RuntimeError(f"Command failed ({result.returncode}): {' '.join(cmd)}")
    return result.stdout


@asset(group_name="bronze")
def bronze_business(context: AssetExecutionContext) -> MaterializeResult:
    """Extract customers, gpus, rentals, prices from Postgres into the bronze layer.
    Wraps ingest/extract_postgres.py from Lab 02."""
    _run(context, ["python", "extract_postgres.py"], cwd=INGEST)
    con = duckdb.connect(str(WAREHOUSE), read_only=True)
    rows = con.execute("select count(*) from bronze.rentals").fetchone()[0]
    con.close()
    return MaterializeResult(metadata={"bronze_rental_rows": MetadataValue.int(rows)})


@asset(group_name="bronze")
def telemetry_bronze(context: AssetExecutionContext) -> MaterializeResult:
    """Drain the Redpanda telemetry topic and land it into bronze.
    Wraps ingest/consume_stream.py from Lab 04."""
    _run(context, ["python", "consume_stream.py", "--drain"], cwd=INGEST)
    con = duckdb.connect(str(WAREHOUSE), read_only=True)
    rows = con.execute("select count(*) from bronze.telemetry").fetchone()[0]
    con.close()
    return MaterializeResult(metadata={"bronze_telemetry_rows": MetadataValue.int(rows)})


@asset(group_name="gold", deps=[bronze_business, telemetry_bronze])
def dbt_build(context: AssetExecutionContext) -> MaterializeResult:
    """Run `dbt build` to produce silver staging + the gold star schema
    (dim_customer, dim_gpu, dim_date, fct_rentals). From Lab 03.
    Depends on BOTH bronze assets — dbt reads the tables they land."""
    _run(context, ["dbt", "build", "--target", "dev"], cwd=DBT_DIR)
    return MaterializeResult(metadata={"dbt": MetadataValue.text("silver + gold built")})


@asset(group_name="gold", deps=[dbt_build])
def fct_telemetry_hourly(context: AssetExecutionContext) -> MaterializeResult:
    """Roll the raw telemetry firehose into an hourly fact.
    Depends on dbt_build because it reads the staged silver telemetry. From Lab 04."""
    con = duckdb.connect(str(WAREHOUSE))
    con.execute("""
        create or replace table gold.fct_telemetry_hourly as
        select
            gpu_id,
            date_trunc('hour', event_ts)      as hour,
            count(*)                           as reading_count,
            avg(utilization_pct)               as avg_utilization,
            max(temperature_c)                 as max_temp_c,
            avg(power_watts)                   as avg_power_watts
        from silver.stg_telemetry
        group by 1, 2
    """)
    rows = con.execute("select count(*) from gold.fct_telemetry_hourly").fetchone()[0]
    con.close()
    return MaterializeResult(metadata={"hourly_rows": MetadataValue.int(rows)})

Two ways to declare a dependency, both shown above:

Style	How	When to use
Argument	Add the upstream asset as a function parameter; Dagster passes its loaded value in.	When the downstream asset uses the upstream's data in-process.
`deps=[...]`	List upstream assets in the decorator; you get ordering but not the value.	When the dependency is "this must run first" — exactly our case, since data flows through DuckDB, not Python.

We use deps=[...] everywhere because every step reads and writes the shared DuckDB file — the data never passes through Python variables, so we only need ordering. Notice dbt_build declares deps=[bronze_business, telemetry_bronze]: it will not start until both bronze assets are green.

Now make Dagster aware of the file. Create dagster_app/definitions.py, which is the entry point Dagster loads:

dagster_app/definitions.py

from dagster import Definitions, load_assets_from_modules

from dagster_app import assets

all_assets = load_assets_from_modules([assets])

defs = Definitions(
    assets=all_assets,
)

And a tiny pyproject.toml stanza (or dagster.yaml) so dagster dev knows which module holds your definitions. The simplest path is a pyproject.toml at the repo root:

pyproject.toml (add this)

[tool.dagster]
module_name = "dagster_app.definitions"
code_location_name = "mini_griddp"

✓ You should see

No errors yet — these are definitions, not runs. Sanity-check the wiring without launching the UI: dagster asset list should print four assets — bronze_business, telemetry_bronze, dbt_build, fct_telemetry_hourly. If it complains it can't find a module, you're likely running from the wrong directory or your package isn't importable — run from the repo root and make sure dagster_app/ has an __init__.py.

Run the graph

Launch Dagster's local development server. It starts a web UI and watches your code for changes:

terminal

# from the repo root, with your venv active
dagster dev

# it prints something like:
#   Serving dagster-webserver on http://127.0.0.1:3000

Open http://127.0.0.1:3000. Click Assets in the top nav, then View global asset lineage (or open the Asset graph). You'll see your four boxes connected by arrows — exactly the dependency graph you declared, drawn for you:

┌────────────────┐ ┌──────────────────┐ │ bronze_business│ │ telemetry_bronze │ (group: bronze) └───────┬────────┘ └────────┬─────────┘ │ │ └──────────┬────────────┘ ▼ ┌──────────────┐ │ dbt_build │ (group: gold) └──────┬───────┘ ▼ ┌──────────────────────┐ │ fct_telemetry_hourly │ (group: gold) └──────────────────────┘

Now run it. Click Materialize all (top-right of the asset graph). Dagster computes the order from the arrows and launches a run. Watch the assets turn from grey to blue (running) to green (materialized), bottom-up the dependency chain.

✓ You should see

The two bronze assets start first and in parallel (they have no dependency on each other). Only once both are green does dbt_build start. Then fct_telemetry_hourly runs last. Every box ends green, and clicking one shows the metadata you returned — bronze_rental_rows, hourly_rows, and so on — plus the streamed subprocess logs. If a box goes red, click it, read the logs (Dagster captured the script's stderr for you), fix, and click Materialize on just that asset.

Re-materialize one asset

You don't have to re-run everything. Click a single asset and hit Materialize to rebuild just it (and, optionally, everything downstream). This is already a huge upgrade over hand-running scripts.

Add a schedule

A graph you click is nice. A graph that runs itself every morning is a platform. Dagster schedules target a job — a named selection of assets to materialize. We'll define a job that covers all four assets and a schedule that runs it daily at 06:00.

dagster_app/definitions.py (updated)

from dagster import (
    Definitions,
    load_assets_from_modules,
    define_asset_job,
    ScheduleDefinition,
    AssetSelection,
)

from dagster_app import assets

all_assets = load_assets_from_modules([assets])

# A job that materializes every asset in the pipeline.
pipeline_job = define_asset_job(
    name="full_pipeline",
    selection=AssetSelection.all(),
)

# Run the whole pipeline daily at 06:00 (cron syntax: min hour dom mon dow).
daily_schedule = ScheduleDefinition(
    name="daily_pipeline",
    job=pipeline_job,
    cron_schedule="0 6 * * *",
    execution_timezone="America/New_York",
)

defs = Definitions(
    assets=all_assets,
    jobs=[pipeline_job],
    schedules=[daily_schedule],
)

Save the file. dagster dev auto-reloads. In the UI, go to Automation (or Schedules), find daily_pipeline, and flip its toggle to on. Schedules only fire while the Dagster daemon is running — and dagster dev runs the daemon for you locally.

✓ You should see

The schedule listed with its cron expression 0 6 * * * and a "Next tick" timestamp for tomorrow at 06:00. Don't want to wait until morning to test it? Use the schedule's ⋯ → Preview / Test schedule, or just click Materialize all on the asset graph to trigger the same job by hand right now.

Why this beats cron-ing the scripts

You might ask: I could just write four lines in crontab and call it a day. You could — and you'd lose almost everything that makes a pipeline operable. This is the practical payoff of Course 4, Chapter 10 (Transformation & Orchestration), made concrete:

Concern	Four cron lines	Dagster asset graph
Ordering	You hard-code `sleep 300` and hope the extract finished. Fragile.	dbt literally cannot start until both bronze assets are green. Order is guaranteed by the graph.
Failure handling	Step 1 fails, steps 2–4 run anyway on stale data. Silent corruption.	A failed upstream asset blocks its downstreams; the run stops at the break, not after it.
Retries	None, unless you script them yourself.	Add `retry_policy=RetryPolicy(max_retries=2)` to an asset — transient blips self-heal.
Visibility	`grep` through log files to learn what happened, if you logged at all.	A UI shows every run, every asset's status, metadata, and captured logs in one place.
Backfills	Re-run yesterday? Hand-edit dates and pray.	Select assets, pick a range, click Backfill. Dagster tracks each partition.
Partial reruns	All-or-nothing.	Re-materialize one asset and its downstreams; leave the rest untouched.

The mental shift

Cron schedules tasks ("run this command at this time"). Dagster manages assets ("this table should exist and be fresh; here's how to make it, and what it needs first"). When you think in assets, the orchestrator can reason about your data — what's stale, what failed, what to rebuild — instead of blindly firing commands into the dark.

Materialize end to end

Let's prove the whole thing runs clean from a cold start. Stop the server (Ctrl-C), confirm the stack is up, and run a fresh materialization — this time from the command line, no UI clicking, the way the scheduler will:

terminal

# make sure Postgres + Redpanda are running
docker compose up -d

# materialize the entire pipeline as one job, headless
dagster job execute -j full_pipeline -m dagster_app.definitions

# ...then peek at the result in DuckDB
duckdb warehouse/griddp.duckdb \
  "select count(*) as hourly_rows from gold.fct_telemetry_hourly;"

✓ You should see

Dagster logs each asset starting and finishing in dependency order — bronze_business and telemetry_bronze first, then dbt_build, then fct_telemetry_hourly — and a final "RUN_SUCCESS". The DuckDB query returns a positive hourly_rows count. That single command just did what used to be four manual steps in the right order. That's the lab.

Commit Lab 05

Your orchestration layer now exists in code. Commit it so your history keeps telling the story:

terminal

cd mini-griddp
git add dagster_app/ pyproject.toml
git status                 # confirm only the orchestration files are staged
git commit -m "Lab 05: Dagster asset graph + daily schedule (ingest → dbt → rollup)"

Don't commit run state

When dagster dev runs, it writes a local $DAGSTER_HOME (or a tmp instance) full of run history and logs. Add it to .gitignore — it's machine state, not source. Your repo should contain the definitions, never the run database.

✓ Check yourself

Can you explain, pointing at the asset graph, why dbt_build can't start until both bronze assets are green?
What's the difference between declaring a dependency as a function argument versus deps=[...] — and why did we use deps here?
Name two things Dagster gives you that four cron lines don't.
Where does the schedule's cron expression 0 6 * * * fire, and what must be running for it to fire at all?

Exercise — Add an asset check (or a downstream asset) and re-materialize

An asset check is a lightweight assertion that runs with an asset and reports pass/fail in the UI — a preview of the data-quality work in Lab 06. Add a check that fails if the hourly rollup came out empty, then re-materialize fct_telemetry_hourly and confirm the check shows up green.

dagster_app/assets.py (add to the bottom)

from dagster import asset_check, AssetCheckResult


@asset_check(asset=fct_telemetry_hourly)
def hourly_rollup_not_empty() -> AssetCheckResult:
    """Fail the build if the hourly fact materialized with zero rows."""
    con = duckdb.connect(str(WAREHOUSE), read_only=True)
    rows = con.execute("select count(*) from gold.fct_telemetry_hourly").fetchone()[0]
    con.close()
    return AssetCheckResult(
        passed=bool(rows > 0),
        metadata={"row_count": MetadataValue.int(rows)},
    )

Save (the server auto-reloads), open fct_telemetry_hourly in the UI, and click Materialize. Under the asset you'll now see a Checks tab with hourly_rollup_not_empty reporting Passed and the row_count metadata.

Prefer a downstream asset instead? Add one that depends on fct_telemetry_hourly — for example a fleaderboard_top_gpus asset that selects the ten busiest GPUs by total readings — declare deps=[fct_telemetry_hourly], and watch it appear as a new box hanging off the bottom of the graph. Re-materialize all and confirm it runs dead last.

Either way, you've extended a live, orchestrated pipeline by adding one declaration — no scheduling logic, no ordering code. That's the power you just unlocked.

Your pipeline runs in order, on a schedule — but nothing yet checks that the data it produces is correct. Let's add real tests and freshness checks. → Lab 06 — Data Quality & Tests