Lab 05 — Orchestrate with Dagster

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:

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.

Make dagster_app/ a package (mkdir -p dagster_app && touch dagster_app/__init__.py) and create dagster_app/assets.py. We'll reuse the exact scripts you already wrote, calling them as subprocesses from the repo root so there's a single source of truth for each step and every relative path (warehouse.duckdb) means what it meant when you ran them by hand. (You could also import and call the functions directly — both are common; subprocess keeps the lab honest about what's actually running.)

import subprocess
import sys
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]
DBT_DIR = REPO / "dbt"
WAREHOUSE = REPO / "warehouse.duckdb"
PYTHON = sys.executable            # the project venv's interpreter


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, "ingest/extract_postgres.py"], cwd=REPO)
    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 in its --drain mode."""
    _run(context, [PYTHON, "ingest/consume_stream.py", "--drain"], cwd=REPO)
    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.
    The telemetry rollup (and anything built on it) is excluded here; that is the next asset."""
    _run(context, ["dbt", "build", "--profiles-dir", ".",
                   "--exclude", "fct_telemetry_hourly+"], 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: the Lab 04 dbt model.
    Depends on dbt_build because it joins the dim_gpu that dbt just rebuilt."""
    _run(context, ["dbt", "run", "--profiles-dir", ".",
                   "--select", "fct_telemetry_hourly+"], cwd=DBT_DIR)
    con = duckdb.connect(str(WAREHOUSE), read_only=True)
    rows = con.execute("select count(*) from gold.fct_telemetry_hourly").fetchone()[0]
    con.close()
    return MaterializeResult(metadata={"hourly_rows": MetadataValue.int(rows)})

The rollup is a dbt model (Lab 04), so the two gold assets split one dbt project by selection. The trailing + is dbt's graph operator for "and everything downstream of it": dbt_build excludes the rollup and its descendants, and the rollup asset runs exactly that set. Without the +, a later model built on the rollup (Lab 07 adds one) would run inside dbt_build and fail on a fresh warehouse, because the rollup it reads does not exist yet. That keeps the rollup its own box in the graph, which is what you want to re-materialize after a telemetry drain without touching the star schema.

Two ways to declare a dependency:

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:

from dagster import (
    Definitions,
    load_assets_from_modules,
    load_asset_checks_from_modules,
    in_process_executor,
)

from dagster_app import assets

all_assets = load_assets_from_modules([assets])
all_checks = load_asset_checks_from_modules([assets])   # @asset_check functions (none yet)

defs = Definitions(
    assets=all_assets,
    asset_checks=all_checks,
    # DuckDB allows one writer per file. Run assets one at a time in this
    # process instead of Dagster's default of one subprocess per asset.
    executor=in_process_executor,
)

And a tiny pyproject.toml stanza so dagster dev knows which module holds your definitions. Add it to the pyproject.toml at the repo root:

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

Run the graph

Launch Dagster's local development server. It starts a web UI and watches your code for changes. The Lab 01 placeholder container is holding port 3000, so stop it first; from now on the real Dagster runs on your laptop:

# from the repo root
docker compose stop dagster     # free port 3000 (the Lab 01 placeholder)
uv run 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:

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.

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.

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

from dagster_app import assets

all_assets = load_assets_from_modules([assets])
all_checks = load_asset_checks_from_modules([assets])

# A job that materializes every asset in the pipeline (and runs its checks).
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,
    asset_checks=all_checks,
    jobs=[pipeline_job],
    schedules=[daily_schedule],
    executor=in_process_executor,   # one DuckDB writer at a time
)

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.

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:

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:

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

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

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

Commit Lab 05

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

cd mini-griddp
printf '.tmp_dagster_home_*/\n' >> .gitignore
git add dagster_app/ pyproject.toml uv.lock .gitignore
git status                 # confirm only the orchestration files are staged
git commit -m "Lab 05: Dagster asset graph + daily schedule (ingest → dbt → rollup)"

✓ Check yourself

Answer from memory before opening each one.

1. Every dependency in assets.py is declared with deps=[...] rather than as a function argument. What would the argument style give you that deps does not, and why is it of no use here?

   Show answer

   The argument style passes the upstream asset's loaded value into the downstream function, in-process. Nothing travels through Python in this pipeline — every step reads and writes the shared DuckDB file — so all that is needed is ordering, and that is exactly what deps provides. See Define the assets.

2. Postgres is down when the 06:00 run starts. Walk through what happens to each of the four assets — and what four cron lines would have done instead.

   Show answer

   bronze_business fails: the subprocess exits non-zero and _run raises. telemetry_bronze does not depend on it, so it still runs. dbt_build needs both bronze assets green and never starts, and fct_telemetry_hourly behind it doesn't either — the run stops at the break. Cron would have fired all four on time regardless, rebuilding gold on stale bronze with nobody told. See Why this beats cron.

3. You toggle daily_pipeline on, close the terminal that was running dagster dev, and go to bed. What happens at 06:00 — and 06:00 in which timezone?

   Show answer

   Nothing. Schedules fire only while the Dagster daemon is running, and locally dagster dev is what runs it; the toggle records intent, but a process has to be alive to act on it. The cron string 0 6 * * * is evaluated in the schedule's execution_timezone, America/New_York — not your laptop's zone and not UTC. See Add a schedule.

4. fct_telemetry_hourly goes red because of a typo in its SQL; the three assets upstream are green. After fixing the typo, what do you run?

   Show answer

   Materialize just that asset (plus anything downstream of it). The upstream assets already materialized and nothing about them changed, so re-running the whole graph would re-extract and rebuild dbt for no benefit. Partial re-runs are one of the things the asset graph buys you over all-or-nothing scripts. See Re-materialize one asset.

5. dbt_build is a single box standing for every dbt model. What do you give up compared with one Dagster asset per model, and when is the single box still the right call?

   Show answer

   Granularity. If one model fails the whole box is red, a retry re-runs every model, and the graph cannot show which table is stale or let you rebuild one mart. For a small project that builds in seconds, one box is simpler and perfectly honest. As models multiply or get slow, per-model assets win — the dagster-dbt integration (@dbt_assets) turns every model into its own asset and every dbt test into an asset check. This capstone keeps the single box. See Define the assets.

6. A teammate has one nightly script with nothing downstream of it and asks why it needs Dagster at all. Where is the line?

   Show answer

   For a single independent step, most of cron's weaknesses don't bite: there is no ordering to get wrong and nothing downstream to poison, so what you forgo is retries and a run history. The line is dependencies. Once a second step reads what the first wrote, you need "don't start until upstream succeeded" and "stop at the break" — which cron can only imitate with sleep and hope. See the comparison table.

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)},
    )

Takeaway

• An orchestrator moves the run order out of your head and into code: four assets, with the two bronze assets first, then dbt_build, then the telemetry rollup.
• Use deps=[...] when data moves through the shared DuckDB file and you need ordering only, and a function argument when the value itself passes in memory.
• A DuckDB file has one writer, and Dagster's default executor runs independent assets at once, so the second bronze asset dies on a file lock; the in-process executor runs one step at a time.
• The trailing + in the dbt selection keeps the rollup and everything downstream out of dbt_build, so a later model built on the rollup cannot fail on a fresh warehouse.
• A job names the assets to materialize and a schedule runs it on a cron expression, but schedules fire only while the Dagster daemon is running.
• Compared with cron, the graph guarantees order, blocks downstream on a failed upstream, supports retries and shows every run in a UI.

The pipeline now runs in order and on a schedule, but nothing checks that its output is correct, so the next lab adds tests and freshness checks. → Lab 06 — Data Quality & Tests

Further reading

• Dagster: Assets — Dagster Labs docs, accessed 2026. Defining assets and declaring upstream dependencies.
• Dagster: Jobs — Dagster Labs docs, accessed 2026. Selecting a set of assets to run together.
• Dagster: Schedules — Dagster Labs docs, accessed 2026. Cron schedules and what the daemon must do.
• Dagster: Run executors — Dagster Labs docs, accessed 2026. Why the in-process executor suits a single-writer warehouse.
• dbt: Node selection syntax — dbt Labs docs, accessed 2026. The + graph operator used to split the dbt project between two assets.