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

• Can you name the four jobs an orchestrator does that cron does not?
• Can you explain the difference between task-based and asset-based orchestration — and why asset-based fits data work?
• In a Dagster asset function, what creates a dependency on an upstream asset?
• When would you use a sensor instead of a schedule?

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

Next

Your pipeline runs itself now — but running on time means nothing if the numbers are wrong. Next we make the data trustworthy: testing and quality checks that catch bad data before anyone sees it. → Data Quality & Testing