Orchestration with Dagster

The pile-of-scripts problem

Most data pipelines start the same innocent way: a few scripts, and a cron entry to run each one on a schedule. extract.py at 1am, transform.py at 2am, load.py at 3am. It works — until it doesn't. And cron will never tell you that it didn't.

Here's what that pile actually looks like, versus what you want:

The cron version has four fatal gaps. There's no dependency ordering — cron just fires at a clock time and prays the upstream job already finished. There are no retries — a one-second network blip means no data for the whole day. There's no visibility — you can't see what ran, what it produced, or how long it took. And worst of all, it fails silently — a script crashes at 2:07am and the first you hear of it is an empty dashboard on Monday morning. An orchestrator closes all four gaps.

What an orchestrator does — the four jobs

Strip away the jargon and an orchestrator has exactly four jobs. Hold these in your head; everything else is detail.

A fifth capability falls out of the first four: backfill — re-running history for a range of dates when you change logic or recover from an outage. We'll touch that at the end.

Dagster's asset model

Most older orchestrators (Airflow is the classic) are task-based: you describe a graph of actions — "run script A, then run script B" — and the tool runs them in order. It never knows or cares what those actions produce. The graph is about doing, not about having.

Dagster flips this. You declare data assets — the things you actually want to exist: a table, a file, a model. Each asset is a Python function that produces that thing, and you say which other assets it depends on. Dagster reads those dependencies and builds the graph for you. You describe the world you want; it figures out the order.

Why does asset-based fit data work so well? Because in data engineering you almost never care that a script ran — you care that a table is fresh and correct. The asset model makes that the first-class thing. Your dependency graph doubles as your data lineage map, for free.

Here's the whole idea in code. An asset is a plain function decorated with @asset; it depends on another asset simply by naming it as a parameter:

from dagster import asset

@asset
def extract_rentals():
    # produce the raw data — return it, and Dagster passes it downstream
    return [{"gpu": "a100", "hours": 3, "rate": 2}, {"gpu": "h100", "hours": 1, "rate": 5}]

@asset
def transform_revenue(extract_rentals):   # ← the PARAMETER name = the upstream asset
    # Dagster sees this dependency and orders the graph automatically
    return [{**r, "revenue": r["hours"] * r["rate"]} for r in extract_rentals]

Notice what you did not write: any ordering, any "run extract before transform" instruction. The parameter extract_rentals on transform_revenue is the dependency. Dagster reads it and knows the order. That's the asset model in one breath.

Schedules & sensors

Declaring the graph is half the job; the other half is when it runs. Dagster gives you two triggers:

• Schedules — time-based, expressed as cron. "Materialize these assets every day at 6am." This is the workhorse for most batch pipelines.
• Sensors — event-based and data-aware. A sensor watches for a condition — a new file landed in a bucket, an upstream table got fresh — and kicks off a run the moment it's true, instead of waiting for the clock. This is how you get pipelines that react to data rather than guessing at timing.

Hands-on: build the marketplace graph

Time to build a real one. We'll create three assets for the GPU-rental marketplace — extract_rentals → transform_revenue → load_mart — run Dagster locally, and watch the lineage graph light up in the UI.

In your mini-griddp repo, create a folder and a single file:

mkdir -p orchestration
cd orchestration

Now the assets. Create orchestration/marketplace.py:

# orchestration/marketplace.py
import duckdb
from dagster import asset, Definitions, MaterializeResult, MetadataValue


@asset
def extract_rentals():
    """Stage 1: pull raw rentals. Here we fake a small batch; in production
    this would read from Postgres (Course 4, ch. 05 ingestion)."""
    rows = [
        {"customer": "acme",  "gpu": "a100", "hours": 3, "rate": 2.0},
        {"customer": "acme",  "gpu": "h100", "hours": 1, "rate": 5.0},
        {"customer": "globex", "gpu": "a100", "hours": 4, "rate": 2.0},
    ]
    return rows


@asset
def transform_revenue(extract_rentals):
    """Stage 2: derive revenue per rental. Depends on extract_rentals
    purely by naming it as a parameter — Dagster orders the graph."""
    return [
        {**r, "revenue": round(r["hours"] * r["rate"], 2)}
        for r in extract_rentals
    ]


@asset
def load_mart(transform_revenue):
    """Stage 3: write a revenue-by-GPU mart into DuckDB."""
    con = duckdb.connect("marketplace.duckdb")
    con.execute("CREATE OR REPLACE TABLE revenue_by_gpu (gpu VARCHAR, revenue DOUBLE)")
    totals = {}
    for r in transform_revenue:
        totals[r["gpu"]] = totals.get(r["gpu"], 0) + r["revenue"]
    for gpu, rev in totals.items():
        con.execute("INSERT INTO revenue_by_gpu VALUES (?, ?)", [gpu, round(rev, 2)])
    n = con.execute("SELECT count(*) FROM revenue_by_gpu").fetchone()[0]
    con.close()
    # surface a number into the UI so you can see what the run produced
    return MaterializeResult(metadata={"rows_written": MetadataValue.int(n)})


# Definitions is how Dagster discovers your assets.
defs = Definitions(assets=[extract_rentals, transform_revenue, load_mart])

Three functions, two implicit dependencies, and one Definitions object that hands the assets to Dagster. That's a complete project. Now launch the local UI from the same folder:

# from inside the orchestration/ folder
dagster dev -f marketplace.py

Dagster starts a local web server (it prints a URL, usually http://127.0.0.1:3000). Open it. Click into Assets → View global asset lineage, then hit the Materialize all button to run the graph.

Backfills & partitions

Two more ideas you'll meet constantly, briefly. So far our assets produce "the latest" data in one lump. Real pipelines usually slice data by time — one chunk per day. In Dagster you declare a partitioned asset (say, daily partitions keyed by date), and each day's data becomes its own independently-materializable cell.

Once an asset is partitioned, a backfill is just "materialize this range of partitions." Changed your revenue logic and need to recompute the last 30 days? Select the date range in the UI and Dagster runs all 30 partitions, in order, with retries and a progress bar — no hand-written loop, no forgotten dates. This is the single feature that makes recovering from an outage or a logic bug a calm afternoon instead of a panic.

How this connects

The three-asset toy you built is the same shape as a production platform — just bigger. In a real mini-griddp, Dagster sits on top of everything you've learned in this course and runs it as one coordinated graph:

• Your ingestion (Course 4, ch. 05–06) becomes the upstream "extract" assets that pull from Postgres.
• Your dbt models (ch. 07) become assets too — Dagster has a native integration that turns each dbt model into a Dagster asset, so your SQL transformations and your Python steps live in one lineage graph.
• Your data-quality tests (next chapter) run as part of the graph — a failing test can block downstream assets, so bad data never reaches a dashboard.

That unified graph — ingest, transform, test, all scheduled and observable in one place — is exactly what you'll assemble in the capstone. You just built its smallest working version.

✓ Check yourself

Answer from memory before opening each one.

1. Under cron, extract.py runs at 1am and transform.py at 2am. One night the source is slow and the extract takes 90 minutes. What does the transform do at 2am, who finds out, and which of the orchestrator's four jobs would have prevented it?

   Show answer

   It runs anyway, on yesterday's or half-written data — cron fires on the clock and hopes the upstream job has finished — and nobody finds out until a dashboard looks wrong. Order prevents it: an orchestrator never starts a step until its inputs are ready. Observe & alert is what would have told you. See The pile-of-scripts problem and the four jobs.

2. A task-based DAG has shown every task green for a week, yet the revenue table hasn't changed in that time. What question was the orchestrator answering, and what question does the asset model put first instead?

   Show answer

   A task graph is about doing: it answers "did the job run?", and never knows or cares what the action produced. The asset model is about having: the unit is "this table exists", so the natural question is "is this table fresh and correct?" — which is the one data work actually cares about, and why the asset graph doubles as lineage. See Dagster's asset model.

3. marketplace.py contains no line saying "run the extract first". How does Dagster know the order — and if you rename the first function to extract_rentals_pg and change nothing else, what happens to the graph?

   Show answer

   The dependency is the parameter name: transform_revenue(extract_rentals) names the upstream asset, and Dagster passes that asset's return value in. After the rename, the parameter points at an asset that no longer exists, so the two are no longer wired together — a rename or typo is a broken graph, and the spelling is the first thing to check. See The dependency is the parameter name.

4. The extract returns three rentals, yet load_mart reports rows_written: 2. Why 2, what are the two rows — and what does it report if you click Materialize all a second time?

   Show answer

   The mart is one row per GPU, not per rental: a100 = 14.0 (3 × 2.0 + 4 × 2.0) and h100 = 5.0. A second run still reports 2, because the asset starts with CREATE OR REPLACE TABLE — re-running it replaces the table rather than appending, which is the idempotent habit from Chapter 05 and what makes retries and backfills safe. See the hands-on.

5. A vendor drops a GPU price file into a bucket at some point between 1am and 9am. One option is a schedule at 9am, another a schedule at 1am, a third a sensor. What does each of the two schedules cost you, and when is a schedule still the right trigger?

   Show answer

   The 9am schedule is always safe but leaves prices up to eight hours stale; the 1am schedule usually runs before the file exists. A sensor watches for the file and starts the run the moment it lands — work that should follow data arrival. A schedule stays right for work that is naturally periodic and doesn't wait on an external arrival, such as a nightly revenue refresh. See Schedules & sensors.

6. You fix a bug in the revenue logic that has affected the last 30 days. Compare the recovery when the asset is one unpartitioned lump with the recovery when it has daily partitions.

   Show answer

   Unpartitioned, your only move is "re-run everything" — slow and risky — or a hand-written loop over dates that is easy to get wrong. With daily partitions each day is its own materializable cell, so the backfill is "materialize these 30 partitions": Dagster runs them with retries and progress tracking, and leaves the unaffected history alone. See Backfills & partitions.

from dagster import (
    asset, Definitions, MaterializeResult, MetadataValue,
    define_asset_job, ScheduleDefinition,
)

@asset
def top_gpu(load_mart):
    """Fourth asset, downstream of load_mart. (load_mart returns a
    MaterializeResult, so read the mart back from DuckDB.)"""
    import duckdb
    con = duckdb.connect("marketplace.duckdb")
    gpu, rev = con.execute(
        "SELECT gpu, revenue FROM revenue_by_gpu ORDER BY revenue DESC LIMIT 1"
    ).fetchone()
    con.close()
    return MaterializeResult(
        metadata={"top_gpu": MetadataValue.text(gpu),
                  "revenue": MetadataValue.float(rev)}
    )

# a job that targets all assets, and a schedule that runs it daily at 6am
daily_job = define_asset_job("daily_refresh", selection="*")
daily_schedule = ScheduleDefinition(job=daily_job, cron_schedule="0 6 * * *")

defs = Definitions(
    assets=[extract_rentals, transform_revenue, load_mart, top_gpu],
    jobs=[daily_job],
    schedules=[daily_schedule],
)

Takeaway

• Cron fires at a clock time with no dependency order, no retries and no visibility, so a failed job simply stays failed.
• An orchestrator does four jobs: schedule, order by dependency, retry and recover, and show every run with alerts.
• Dagster is asset-based: you declare the tables and files that should exist, and the graph comes from each function's parameters, so it doubles as lineage.
• A misspelled upstream parameter name is a broken graph, not a typo that runs, so check the lineage view when a dependency is missing.
• Use a schedule for naturally periodic work and a sensor when work should follow data arriving.
• Partition an asset by time and a backfill becomes selecting a range, which re-runs only the slice that changed.
• dbt models become assets and quality tests can sit in the graph, so one graph can block bad data before it reaches a dashboard.

The pipeline now runs itself, but running on time means nothing if the numbers are wrong, so the next chapter makes the data trustworthy. → Data Quality & Testing

Further reading

• Dagster: Assets — Dagster Labs docs, accessed 2026. The asset model, dependencies and materialization.
• Dagster: Schedules — Dagster Labs docs, accessed 2026. Cron-style schedules over assets and jobs.
• Dagster: Sensors — Dagster Labs docs, accessed 2026. Event-driven runs that react to new data.
• Dagster: Partitions and backfills — Dagster Labs docs, accessed 2026. Time-sliced assets and how a backfill selects a range.
• Dagster: dbt integration — Dagster Labs docs, accessed 2026. Loading a dbt project so each model is an asset.