Asset-Oriented Orchestration: Dagster and Prefect

Consider a hypothetical retail pipeline that builds a daily_sales table, a trailing seven-day total from it, and a weekly report from that total. On October 8 the source team corrects some orders from October 2. Three questions follow. Which outputs are now wrong? Which version of October 2 did last Friday’s report read? And if tomorrow’s rebuild of October 2 silently loses a store, what keeps the incomplete table away from readers?

A task-centered scheduler can say that extract_orders succeeded on October 2. None of the three questions is about a task. They are about data objects, the versions of those objects, and the dependencies between them. Asset-oriented orchestration makes the data object the unit of definition: each asset is declared with the assets it reads and the code that produces it, and the orchestrator keeps production records, partition status, and checks attached to the object instead of to whichever job happened to run. Dagster is organized around this model. Prefect adds asset tracking to its task and flow model. Airflow’s documentation describes asset events that tasks emit through declared outlets.

This article builds the mechanism from its parts: what a declared graph lets a planner compute, what a production record can and cannot prove, how a partition mapping decides what to rebuild after a correction, and how a check gates publication. It assumes the DAG, task, and data-interval vocabulary from Apache Airflow Fundamentals and the retry and backfill mechanics from Airflow Operations. The dependency graph that Transformation as Code derives from SQL ref() calls is the same idea applied to SELECT statements.

The programs ran on Python 3.12.14 with dagster 1.13.25, prefect 3.8.7, and the standard library’s sqlite3. All data is synthetic and written for this article, and the outputs shown are copied from those runs. Dagster and Prefect are called in process, so schedulers, daemons, the web UI, and remote storage are not part of the examples.

From a task graph to an asset graph

In a task graph the nodes are actions such as extract, load, and validate, and an edge means “run after”. What each action reads and writes is hidden inside its code. An asset graph changes what a node is. Each node is a data object with a name, such as the table daily_sales, and an edge means “reads”. The function that produces the object belongs to the node. Run order becomes a consequence of the reads instead of something an author sequences by hand.

Dagster’s documentation describes an asset as an object in persistent storage, such as a table, a file, or a trained model, and an asset definition as the code that produces and updates it. The unit is a data object, not a table in particular. A file in object storage, a model artifact, or a dashboard extract can all be assets, and choosing the granularity (one asset per table, per partition, or per file) is a design decision with consequences that appear later in this article.

The practical gain is that questions about data become graph queries. The following program is a standard-library stand-in for the planner inside such a tool.

from collections import deque

# Declared data dependencies: asset -> the assets it reads.
reads = {
    "raw_orders":    [],
    "daily_sales":   ["raw_orders"],
    "sales_7d":      ["daily_sales"],
    "sales_report":  ["sales_7d"],
    "store_ranking": ["daily_sales"],
}

def downstream(reads, changed):
    """Every asset that can be affected by a change, found by walking the edges backwards."""
    readers = {a: [b for b, deps in reads.items() if a in deps] for a in reads}
    seen, queue = {changed}, deque([changed])
    while queue:
        for nxt in readers[queue.popleft()]:
            if nxt not in seen:
                seen.add(nxt); queue.append(nxt)
    return seen

def order(reads, subset):
    """Topological order restricted to a subset: an asset runs after the subset members it reads."""
    done, plan = set(), []
    while len(plan) < len(subset):
        for a in sorted(subset - done):
            if all(d in done or d not in subset for d in reads[a]):
                plan.append(a); done.add(a); break
    return plan

print("declared:", order(reads, downstream(reads, "raw_orders")))
print("daily_sales changed ->", order(reads, downstream(reads, "daily_sales")))

# A job that reads daily_sales without declaring it is invisible to the plan.
reads_missing_edge = {**reads, "finance_export": []}      # really reads daily_sales, edge not declared
plan = order(reads_missing_edge, downstream(reads_missing_edge, "daily_sales"))
print("finance_export is in the plan:", "finance_export" in plan)

The dictionary reads maps each asset to the assets it reads. downstream walks those edges backward from a changed asset, breadth first, and returns every asset that could be affected. order sorts that subset so each asset runs after the members of the subset that it reads; ties break alphabetically so the output is deterministic.

declared: ['raw_orders', 'daily_sales', 'sales_7d', 'sales_report', 'store_ranking']
daily_sales changed -> ['daily_sales', 'sales_7d', 'sales_report', 'store_ranking']
finance_export is in the plan: False

Rebuilding from raw_orders yields all five assets in dependency order. A change to daily_sales skips raw_orders and plans only the three assets that read it, directly or transitively. Nobody wrote either sequence; both come from the declared reads. The third line shows the limit of this approach. finance_export really reads daily_sales, but nobody declared that edge, so the plan never includes it, and it will keep serving numbers computed from the old table until someone notices. The planner is exactly as complete as the declarations, and no orchestrator can infer an undeclared read from the body of a Python function.

Airflow has assets as well. Its documentation for 3.3.2 says a task that lists outlets=[Asset(...)] produces an asset event when it succeeds, and that a DAG can use schedule=[Asset(...)] to run after those assets update. In that design the asset is a signal between workflows that are still defined as tasks. In Dagster the asset is the definition itself. The rest of this article exercises Dagster and Prefect, whose asset APIs the examples call directly.

Dagster: an asset is code plus a production record

In Dagster a function decorated with @dg.asset defines an asset. The function name is the asset key, deps names the assets it reads, partitions_def slices it into partitions, and code_version labels the logic. Returning MaterializeResult(metadata=...) attaches facts to the event that records the production. dg.materialize runs the selected assets in process and records one event per asset.

The asset below builds one day of daily_sales into a SQLite table. The key line is the table name. It contains the day and a digest of the source rows read for that day, so every distinct input produces a distinct physical object.

import hashlib, os, sqlite3, tempfile
import dagster as dg

DB = os.path.join(tempfile.mkdtemp(), "warehouse.db")
days = dg.DailyPartitionsDefinition(start_date="2024-10-01", end_date="2024-10-10")

def db():
    return sqlite3.connect(DB)

def input_revision(day):
    """Digest of the source rows the asset will read for this day."""
    rows = db().execute("SELECT store, amount FROM orders_src WHERE day=? ORDER BY store, amount", (day,)).fetchall()
    return hashlib.sha256(repr(rows).encode()).hexdigest()[:8]

@dg.asset(partitions_def=days, code_version="1")
def daily_sales(context: dg.AssetExecutionContext):
    day, rev = context.partition_key, input_revision(context.partition_key)
    table = f"daily_sales__{day.replace('-', '')}__{rev}"            # physical version of this partition
    con = db()
    with con:
        con.execute(f"DROP TABLE IF EXISTS {table}")
        con.execute(f"CREATE TABLE {table} AS SELECT store, SUM(amount) AS total FROM orders_src WHERE day=? GROUP BY store", (day,))
    stores, total = con.execute(f"SELECT COUNT(*), SUM(total) FROM {table}").fetchone()
    return dg.MaterializeResult(metadata={"output_table": table, "input_revision": rev,
                                          "stores": stores, "total": total,
                                          "validation": "4 of 4 expected stores" if stores == 4 else f"{stores} of 4 expected stores"})

The metadata is the evidence the event will carry: which physical table was written, which input revision it was computed from, how many stores and what total it contains, and a validation result. Dagster adds the partition key and the code version itself.

What a production record has to contain

If a date is rebuilt, the logical asset name and partition key stay the same while the physical version changes. A consumer that needs reproducibility therefore needs more than the name. It needs the partition key, a reference to the physical output, the input revision, the code version, and the validation result, all retained with the production record. The next block seeds a source table, builds October 2, lets a report read the table the record points to, corrects the source, rebuilds, and then compares what “latest” and “recorded” return.

def records(inst, day):
    flt = dg.AssetRecordsFilter(asset_key=dg.AssetKey("daily_sales"), asset_partitions=[day])
    return inst.fetch_materializations(flt, limit=10, ascending=True).records

def show(inst, day):
    for i, r in enumerate(records(inst, day), 1):
        m = r.asset_materialization
        print(f"  materialization {i}: " + ", ".join(f"{k}={v.value}" for k, v in m.metadata.items() if k != "stores")
              + f", code_version={m.tags['dagster/code_version']}")

con = db()
con.execute("CREATE TABLE orders_src(day TEXT, store TEXT, amount INT)")
con.executemany("INSERT INTO orders_src VALUES (?,?,?)",
                [("2024-10-02", s, a) for s, a in [("S1", 120), ("S2", 80), ("S3", 45), ("S4", 60)]])
con.commit()

with dg.DagsterInstance.ephemeral() as inst:
    dg.materialize([daily_sales], instance=inst, partition_key="2024-10-02")
    print("first build of 2024-10-02")
    show(inst, "2024-10-02")
    first_table = records(inst, "2024-10-02")[-1].asset_materialization.metadata["output_table"].value

    # A report reads the version it was built from and remembers which one that was.
    report_total = db().execute(f"SELECT SUM(total) FROM {first_table}").fetchone()[0]
    print("report built from", first_table, "->", report_total)

    # The source is corrected: one order for S3 changes from 45 to 95. The partition key is unchanged.
    con.execute("UPDATE orders_src SET amount=95 WHERE day='2024-10-02' AND store='S3'"); con.commit()
    dg.materialize([daily_sales], instance=inst, partition_key="2024-10-02")
    print("after the correction and a rebuild")
    show(inst, "2024-10-02")
    latest = records(inst, "2024-10-02")[-1].asset_materialization.metadata["output_table"].value
    print("latest version      ->", db().execute(f"SELECT SUM(total) FROM {latest}").fetchone()[0], latest)
    print("recorded version    ->", db().execute(f"SELECT SUM(total) FROM {first_table}").fetchone()[0], first_table)

    # Retention: the record outlives the table it points to.
    con.execute(f"DROP TABLE {first_table}"); con.commit()
    try:
        db().execute(f"SELECT SUM(total) FROM {first_table}")
    except sqlite3.OperationalError as e:
        print("after retention drops the old table:", e)

first build of 2024-10-02
  materialization 1: output_table=daily_sales__20241002__eed8fe9c, input_revision=eed8fe9c, total=305, validation=4 of 4 expected stores, code_version=1
report built from daily_sales__20241002__eed8fe9c -> 305
after the correction and a rebuild
  materialization 1: output_table=daily_sales__20241002__eed8fe9c, input_revision=eed8fe9c, total=305, validation=4 of 4 expected stores, code_version=1
  materialization 2: output_table=daily_sales__20241002__4253beaa, input_revision=4253beaa, total=355, validation=4 of 4 expected stores, code_version=1
latest version      -> 355 daily_sales__20241002__4253beaa
recorded version    -> 305 daily_sales__20241002__eed8fe9c
after retention drops the old table: no such table: daily_sales__20241002__eed8fe9c

Two events now exist for the same asset key and the same partition. Their output_table values differ because their input revisions differ (eed8fe9c for the original rows, 4253beaa after S3’s order changed from 45 to 95), and the totals are 305 and 355. A report built from the first build can be reproduced by resolving the recorded table. A report that reads “the latest table for October 2” returns 355 and no longer matches what it showed last week. The partition key 2024-10-02 is identical in both events, so it cannot serve as a snapshot identifier. Only the input revision separates them.

The final line of that block shows the limit of a record. After retention drops the old table, the event still says where the report’s input lived, and the table is gone. A record is evidence that production happened, not storage. Reproducibility therefore needs a retention rule: keep a physical version for as long as any published result that cites it must remain reproducible.

Dagster records part of this itself. The next program materializes an upstream partition with a version string supplied by the code and a downstream asset that reads it, then prints the tags on the downstream event.

import dagster as dg

days = dg.DailyPartitionsDefinition(start_date="2024-10-01", end_date="2024-10-10")
SOURCE_REVISION = {"value": "rev-a"}

@dg.asset(partitions_def=days, code_version="1")
def orders_extract(context: dg.AssetExecutionContext):
    return dg.MaterializeResult(data_version=dg.DataVersion(SOURCE_REVISION["value"]))

@dg.asset(partitions_def=days, deps=[orders_extract], code_version="1")
def daily_sales(context: dg.AssetExecutionContext):
    pass

with dg.DagsterInstance.ephemeral() as inst:
    dg.materialize([orders_extract, daily_sales], instance=inst, partition_key="2024-10-02")
    record = inst.fetch_materializations(dg.AssetRecordsFilter(asset_key=dg.AssetKey("daily_sales")), limit=1).records[0]
    for key, value in sorted(record.asset_materialization.tags.items()):
        print(f"{key:<36} {value[:12]}")

dagster/code_version                 1
dagster/data_version                 86ad91c0bf4f
dagster/input_data_version/orders_extract 162af81f5de4
dagster/input_event_pointer/orders_extract 10

The downstream event carries its own dagster/code_version, a dagster/data_version, and for each upstream a digest of the data version it read and a pointer to the upstream materialization event (here event 10 in this run’s log). Dagster’s documentation says data versions are by default a hash of the code version together with the data versions of the inputs, and that code can supply its own. It also says an asset is shown as unsynced when its code version changes, when an upstream data version changes, or when its dependencies change. Compared with the hand-made record above, Dagster’s tags answer “which upstream version did this event read” without any code in the asset. They do not know your physical table names or your validation results, so those still belong in the metadata, as the previous block did.

A materialization event is not a commit

An event is emitted when the step reports its result. The write it describes happened earlier, in code the orchestrator does not control. The final block makes that visible by committing a row and then failing before the step reports anything.

# The write commits, then the step fails before it reports anything.
@dg.asset(partitions_def=days)
def late_failure(context: dg.AssetExecutionContext):
    with db() as c:
        c.execute("CREATE TABLE IF NOT EXISTS late_failure(day TEXT PRIMARY KEY)")
        c.execute("INSERT OR REPLACE INTO late_failure VALUES (?)", (context.partition_key,))
    raise ConnectionError("connection lost after the commit")

with dg.DagsterInstance.ephemeral() as inst:
    r = dg.materialize([late_failure], instance=inst, partition_key="2024-10-03", raise_on_error=False)
    flt = dg.AssetRecordsFilter(asset_key=dg.AssetKey("late_failure"))
    print("run success:", r.success, "| materialization events:", len(inst.fetch_materializations(flt, limit=5).records),
          "| rows in destination:", db().execute("SELECT COUNT(*) FROM late_failure").fetchone()[0])

run success: False | materialization events: 0 | rows in destination: 1

The run failed and Dagster recorded no materialization event, yet the destination holds the row. Absence of an event does not prove absence of data. The opposite gap also exists: an asset that reports success before its destination commit, or one whose result lives only in memory, leaves an event without a durable object. An ephemeral instance, which these programs use, discards its event log when the block exits, so what it records is a demonstration and not an audit trail. For these reasons an event should be read as a record of a production attempt. A statement that a consumer-visible version exists comes from the destination. Where possible, make the commit and its evidence one operation, as in the pattern from Building Incremental and Idempotent Pipelines, and design the write so that a repeat after a lost acknowledgment replaces the row instead of duplicating it (see idempotency). Reconcile event history against the destination periodically instead of trusting either alone.

Partitions and partition mappings

A partitioned asset is a family of slices of one asset. DailyPartitionsDefinition produces keys such as 2024-10-02, and each key has its own materialization history. The slice plays the role the data interval plays in Airflow (see Data Interval and Logical Date): work is named by the slice of data it covers, not by the wall-clock time of the run, so a rebuild of October 2 means the same thing on any day.

When an upstream and a downstream asset share a partitions definition, Dagster’s default maps each downstream partition to the upstream partition with the same key. For time-window partitions, its documentation says a downstream partition depends on every upstream partition that intersects its window; a daily asset built from an hourly one reads that day’s 24 hourly partitions. A trailing window needs an explicit mapping.

A trailing seven-day total for day \(d\) reads the daily values for days \(d-6\) through \(d\). TimeWindowPartitionMapping(start_offset=-6, end_offset=0) states that: the offsets shift the start and end of the upstream window relative to the downstream partition, counted in partitions. The output for October 7 reads October 1 through October 7.

The question after a correction runs the other way. Fix an upstream day \(u\) and ask which outputs read it. Output \(d\) reads \(u\) exactly when \(d-6 \le u \le d\), which is the same condition as \(u \le d \le u+6\). A correction to October 2 therefore reaches the outputs for October 2 through October 8, seven of them. October 1 is unaffected because its window ends on October 1, and October 9 is unaffected because its window begins on October 3. The corrected input date is only the first of seven outputs, and rebuilding only that one leaves six outputs computed from the old value. This reverse traversal finds what to rebuild, and it is wider than the corrected date.

The first block asks Dagster’s mapping object both questions directly. It also defines two toy assets: a daily value for each of nine days and a seven-day sum that records how many days it actually read.

from datetime import datetime
import dagster as dg

days = dg.DailyPartitionsDefinition(start_date="2024-10-01", end_date="2024-10-10", timezone="UTC")
window = dg.TimeWindowPartitionMapping(start_offset=-6, end_offset=0)

SOURCE = {f"2024-10-{d:02d}": 100 + d for d in range(1, 10)}        # toy daily totals, 2024-10-01 .. 09

@dg.asset(partitions_def=days)
def daily_sales(context: dg.AssetExecutionContext) -> int:
    return SOURCE[context.partition_key]

@dg.asset(partitions_def=days, ins={"daily_sales": dg.AssetIn(partition_mapping=window)})
def sales_7d(context: dg.AssetExecutionContext, daily_sales) -> dict:
    keys = context.asset_partition_keys_for_input("daily_sales")
    values = daily_sales if len(keys) > 1 else {keys[0]: daily_sales}   # one mapped partition arrives unwrapped
    return {"total": sum(values.values()), "days_read": len(values), "complete": len(values) == 7}

def keys(subset):
    return sorted(subset.get_partition_keys())

# 1. Forward mapping: which upstream partitions does one output read?
now = datetime(2024, 10, 12, 12, 0)
up = window.get_upstream_mapped_partitions_result_for_partitions(
    days.subset_with_partition_keys(["2024-10-07"]), days, days, current_time=now)
print("2024-10-07 reads", keys(up.partitions_subset))

# 2. Reverse traversal: which outputs does a correction to 2024-10-02 reach?
down = window.get_downstream_partitions_for_partitions(
    days.subset_with_partition_keys(["2024-10-02"]), days, days, current_time=now)
affected = keys(down)
print("a correction to 2024-10-02 reaches", affected)
early = window.get_downstream_partitions_for_partitions(
    days.subset_with_partition_keys(["2024-10-02"]), days, days, current_time=datetime(2024, 10, 5, 12, 0))
print("...if only 2024-10-01 to 04 exist yet:", keys(early))

The current_time argument matters because a daily partition exists only after its day has ended. The second call uses a time on October 5 to show the effect of the right edge: with only October 1 through 4 available, the reverse traversal of a correction to October 2 can reach only October 2 through 4.

The second block then runs the experiment. It builds the history, corrects October 2 in the source, rebuilds every output to see which values really change, and separately rebuilds only the outputs the reverse traversal named.

def build_all(inst, partitions):
    out = {}
    for p in partitions:
        r = dg.materialize([daily_sales, sales_7d], instance=inst, partition_key=p, selection=[sales_7d])
        out[p] = r.output_for_node("sales_7d")
    return out

with dg.DagsterInstance.ephemeral() as inst:
    for p in SOURCE:                                    # initial history
        dg.materialize([daily_sales], instance=inst, partition_key=p)
    before = build_all(inst, SOURCE)

    SOURCE["2024-10-02"] += 50                          # the correction
    dg.materialize([daily_sales], instance=inst, partition_key="2024-10-02")
    after = build_all(inst, SOURCE)                     # rebuild every output, to see what really changed
    targets = [p for p in affected if p in SOURCE]      # restricted rebuild: only the reverse-traversal result
    partial = {**before, **build_all(inst, targets)}

changed = [p for p in SOURCE if before[p]["total"] != after[p]["total"]]
print("outputs whose value changed:", changed)
print("matches the reverse traversal:", changed == [p for p in affected if p in SOURCE])
for p in ("2024-10-01", "2024-10-03", "2024-10-07"):
    print(p, after[p])
print("restricted rebuild of", len(targets), "outputs equals the full rebuild of", len(SOURCE), ":", partial == after)

2024-10-07 reads ['2024-10-01', '2024-10-02', '2024-10-03', '2024-10-04', '2024-10-05', '2024-10-06', '2024-10-07']
a correction to 2024-10-02 reaches ['2024-10-02', '2024-10-03', '2024-10-04', '2024-10-05', '2024-10-06', '2024-10-07', '2024-10-08']
...if only 2024-10-01 to 04 exist yet: ['2024-10-02', '2024-10-03', '2024-10-04']
outputs whose value changed: ['2024-10-02', '2024-10-03', '2024-10-04', '2024-10-05', '2024-10-06', '2024-10-07', '2024-10-08']
matches the reverse traversal: True
2024-10-01 {'total': 101, 'days_read': 1, 'complete': False}
2024-10-03 {'total': 356, 'days_read': 3, 'complete': False}
2024-10-07 {'total': 778, 'days_read': 7, 'complete': True}
restricted rebuild of 7 outputs equals the full rebuild of 9 : True

The forward mapping returns the seven days from October 1, and the reverse traversal returns October 2 through 8. With only four partitions existing, the traversal shrinks to October 2 through 4. The measured result agrees: the outputs whose values changed after the rebuild are exactly the seven that the traversal predicted, and rebuilding those seven reproduces the full rebuild of nine. This is the intended use of the mapping, planning work from declared dependencies and restricting execution to the affected range. The restriction has two parts, the traversal itself and the outputs one intends to restate. A team may decide, for instance, not to restate partitions that were already published to an external party, even though the traversal reaches them.

The last two output lines show a quieter effect. The outputs for October 1 and October 3 read one and three days, and the mapping did not raise or flag that. It returned the part of the window that exists. A consumer comparing the October 3 value of 356, a three-day sum, with the October 7 value of 778, a seven-day sum, compares unlike quantities. Dagster’s mapping did not report the shortfall in these runs, so the asset records days_read and complete itself. Missing prior history must be handled in the asset or in a check, by failing, by marking the output partial, or by starting the output range after the first complete window.

Two further boundaries belong to the declaration. The partitions definition carries a time zone, and an order belongs to a day according to the timestamp’s date in that zone; mapping between partition sets with different zones or granularities moves that boundary. And a window mapping describes dependency scope, not the layout of storage: the sum of seven daily partitions can read one table that is partitioned by month.

Rebuilding a partition needs a safe destination write as well. The write must replace the partition’s complete contents in one transaction and give the same result when repeated; partition overwrite and atomic publication cover those rules. Dagster’s documentation says a backfill over N partitions launches N runs by default, and that BackfillPolicy.single_run() runs a range in one run for engines such as Spark or Snowflake, provided the asset’s code reads the partition range instead of a single key. Whichever is used, a data backfill that finishes after a newer correction was published can overwrite it with older input, and the orchestrator does not prevent that. Late-arriving rows raise the same question from the other side; Late Data and Lookback describes how far back a rebuild must reach.

Asset checks: evidence and gates

A check evaluates one named property of an asset and returns a result: passed or failed, a severity, and metadata describing what was measured. The useful property of a check is that it is evidence about a specific version of the data. The mistake it prevents is treating a green run as proof of the data.

Suppose an asset contains only nonnegative sales but omits an entire store. A range check passes while a completeness check fails, because the range check examines the rows that exist and the missing store has none. Completeness needs an independent expected list. If the list of expected stores were derived from the candidate itself, the check would compare the data with itself and could never fail. In the program below the list lives in a separate store_master table.

import hashlib, os, sqlite3, tempfile
import dagster as dg

DB = os.path.join(tempfile.mkdtemp(), "warehouse.db")
INPUT = {"rows": []}                       # stands in for the day's source rows

def db():
    return sqlite3.connect(DB)

with db() as con:                          # independent list of stores that must report
    con.execute("CREATE TABLE store_master(store TEXT)")
    con.executemany("INSERT INTO store_master VALUES (?)", [("S1",), ("S2",), ("S3",), ("S4",)])

def validate_candidate():
    """One predicate per property. Returns measured values, not just a verdict."""
    con = db()
    expected = {r[0] for r in con.execute("SELECT store FROM store_master")}
    stores = {r[0] for r in con.execute("SELECT store FROM sales_candidate")}
    negative = con.execute("SELECT COUNT(*) FROM sales_candidate WHERE total < 0").fetchone()[0]
    digest = hashlib.sha256(repr(sorted(con.execute("SELECT * FROM sales_candidate"))).encode()).hexdigest()[:8]
    return {"negative_rows": negative, "missing_stores": sorted(expected - stores), "candidate_digest": digest}

@dg.asset
def candidate_sales():
    """Build into a staging table that readers never query."""
    with db() as con:
        con.execute("DROP TABLE IF EXISTS sales_candidate")
        con.execute("CREATE TABLE sales_candidate(store TEXT, total INTEGER)")
        con.executemany("INSERT INTO sales_candidate VALUES (?,?)", INPUT["rows"])

@dg.asset_check(asset=candidate_sales)
def no_negative_totals():
    v = validate_candidate()
    return dg.AssetCheckResult(passed=v["negative_rows"] == 0,
                               metadata={"input_version": v["candidate_digest"], "measured": v["negative_rows"], "threshold": 0})

def completeness(severity):
    def check():
        v = validate_candidate()
        return dg.AssetCheckResult(passed=not v["missing_stores"], severity=severity,
                                   metadata={"input_version": v["candidate_digest"], "measured": str(v["missing_stores"]),
                                             "threshold": "no missing store"})
    return check

stores_error = dg.asset_check(asset=candidate_sales, name="all_stores_present", blocking=True)(completeness(dg.AssetCheckSeverity.ERROR))
stores_warn = dg.asset_check(asset=candidate_sales, name="all_stores_present", blocking=True)(completeness(dg.AssetCheckSeverity.WARN))

def promote():
    with db() as con:
        con.execute("CREATE TABLE IF NOT EXISTS sales_live(store TEXT, total INTEGER)")
        con.execute("DELETE FROM sales_live")
        con.execute("INSERT INTO sales_live SELECT * FROM sales_candidate")

@dg.asset(deps=[candidate_sales])
def publish_unguarded():
    promote()

@dg.asset(deps=[candidate_sales])
def publish_guarded():
    v = validate_candidate()                # the same predicate, evaluated where publication happens
    if v["negative_rows"] or v["missing_stores"]:
        raise Exception(f"candidate {v['candidate_digest']} refused: {v}")
    promote()

validate_candidate measures: the count of negative rows, the stores in the master list but not in the candidate, and a digest of the candidate. Each check wraps that measurement and returns AssetCheckResult with the digest as input_version, the measured value, and the threshold. Together with the pass or fail result, those four items are the record that makes a later question about “what did the check actually see” answerable.

candidate_sales builds into a staging table that readers never query, and publish_unguarded promotes it to the live table in one transaction. Two copies of the completeness check differ only in severity. blocking=True is what asks Dagster to keep downstream assets in the same run from materializing when the check fails. Dagster’s documentation says checks run after the asset’s materialization by default and that blocking prevents downstream materialization in the run; the severity is part of the result.

The scenario helpers (not shown) run the four definitions with an optional asset selection and print the run status, the steps that materialized, the failed checks, and the live table. reset first publishes a complete candidate so that each scenario starts from a known live table.

COMPLETE = [("S1", 120), ("S2", 80), ("S3", 45), ("S4", 60)]
S3_MISSING = [("S1", 120), ("S2", 80), ("S4", 60)]

def reset():
    INPUT["rows"] = COMPLETE
    run("", stores_error, publish_unguarded, quiet=True)      # publish a complete candidate
    INPUT["rows"] = S3_MISSING

reset(); INPUT["rows"] = COMPLETE
run("1. complete candidate", stores_error, publish_unguarded)
reset()
run("2. S3 missing, blocking check with ERROR severity", stores_error, publish_unguarded, evidence=True)
reset()
run("3. S3 missing, blocking check with WARN severity", stores_warn, publish_unguarded)
reset()
run("4a. stage the bad candidate; its checks run", stores_error, publish_unguarded, selection=[candidate_sales])
run("4b. launch the publish asset alone, no guard of its own", stores_error, publish_unguarded, selection=[publish_unguarded])
reset()
run("5a. stage the bad candidate; its checks run", stores_error, publish_guarded, selection=[candidate_sales])
run("5b. launch the publish asset alone, with its own guard", stores_error, publish_guarded, selection=[publish_guarded])

1. complete candidate
    run succeeded=True  materialized=['candidate_sales', 'publish_unguarded']  failed checks=[]
    live table=[('S1', 120), ('S2', 80), ('S3', 45), ('S4', 60)]
2. S3 missing, blocking check with ERROR severity
    run succeeded=False  materialized=['candidate_sales']  failed checks=['all_stores_present(ERROR)']
    live table=[('S1', 120), ('S2', 80), ('S3', 45), ('S4', 60)]
    evidence: all_stores_present False {'input_version': '1664bddc', 'measured': "['S3']", 'threshold': 'no missing store'}
    evidence: no_negative_totals True {'input_version': '1664bddc', 'measured': 0, 'threshold': 0}
3. S3 missing, blocking check with WARN severity
    run succeeded=True  materialized=['candidate_sales', 'publish_unguarded']  failed checks=['all_stores_present(WARN)']
    live table=[('S1', 120), ('S2', 80), ('S4', 60)]
4a. stage the bad candidate; its checks run
    run succeeded=False  materialized=['candidate_sales']  failed checks=['all_stores_present(ERROR)']
    live table=[('S1', 120), ('S2', 80), ('S3', 45), ('S4', 60)]
4b. launch the publish asset alone, no guard of its own
    run succeeded=True  materialized=['publish_unguarded']  failed checks=[]
    live table=[('S1', 120), ('S2', 80), ('S4', 60)]
5a. stage the bad candidate; its checks run
    run succeeded=False  materialized=['candidate_sales']  failed checks=['all_stores_present(ERROR)']
    live table=[('S1', 120), ('S2', 80), ('S3', 45), ('S4', 60)]
5b. launch the publish asset alone, with its own guard
    run succeeded=False  materialized=[]  failed checks=[]
    live table=[('S1', 120), ('S2', 80), ('S3', 45), ('S4', 60)]

Scenario 1 passes both checks, so publish_unguarded runs and the live table has four stores. In scenario 2 store S3 is missing. The range check passes and the completeness check fails with ERROR severity, and the evidence line shows the digest it examined, the missing store, and the threshold. The run fails, publish_unguarded does not run, and the live table keeps its previous four stores. Note that candidate_sales is still listed as materialized: the check ran after its upstream asset had already been produced. Blocking stops downstream work in the run; it does not undo the candidate or revoke access to it. That is why the candidate goes to a staging table.

Scenario 3 changes only the severity. The same failure at WARN severity does not stop the run: publish_unguarded runs and the live table loses S3, even though blocking=True is set. A blocking check enforces a rule only when its failure is an error. The decision for each check is whether a failed property is a reason to stop publication or a reason to alert someone; a completeness check against a master list is usually the former, and a statistical drift check on a metric the business expects to move may be the latter. Scenarios 2 and 3 are the deliberately failing candidates that test the decision, and the same input should be kept as a permanent test case so a change in severity cannot pass unnoticed.

Scenarios 4 and 5 test the scope of the gate. In 4a the bad candidate is staged by launching candidate_sales; Dagster includes that asset’s checks in the selection, the completeness check fails, and the live table is untouched. In 4b an operator, schedule, or sensor launches only the publish asset. No run contains the check, so nothing blocks, and the live table now contains three stores. The blocking relationship is enforced between steps of one run. Selecting a downstream asset by itself, or a check-free job, bypasses it.

Scenario 5 puts the gate where the effect happens. publish_guarded evaluates the same predicate on the table it is about to promote and refuses. Launched alone against the same bad candidate (5b), it fails and the live table keeps its four stores. This guard validates the object it publishes instead of trusting a verdict recorded earlier, so a candidate that changed between the check and the publish cannot slip through. The asset check still has a job: it records evidence for operators and gives the UI and alerts a named result. The enforcement lives with the publication.

Checks have limits beyond scope. A check runs a predicate at one moment against one object, and a passing result says nothing about properties it did not measure. Checks on partitioned assets need care: in this Dagster version, giving a check its own partitions_def emits a PreviewWarning, so the examples check an unpartitioned candidate. Dagster’s documentation also shows checks being selected or excluded separately from assets, which is useful for scheduling them independently and makes skipping them easy. The deeper treatment of what to check and how to reconcile against a trusted reference is in Data Quality Invariants and Business Reconciliation.

Prefect: assets attached to flows

Prefect starts from a different place. A flow is a Python function, and its documentation says the orchestration graph emerges at runtime from the way the function calls tasks and nested flows. There is no graph definition to analyze before the run. Prefect’s asset support attaches asset identity to tasks: @materialize marks a task as producing an asset identified by a key, usually a URI. Lineage comes from two sources, outputs of one materializing task passed to another, and an explicit asset_deps list for dependencies the data flow does not show. The documentation describes the decorator as recording an intent to materialize, and emits prefect.asset.materialization.succeeded or ...failed events according to the task’s final state, plus prefect.asset.referenced events for upstream assets.

The program has two flows that build the same asset. The first caches build_by_day on its inputs, so the cache key is the day. The second adds the source revision to the inputs. Both use INPUTS, one of Prefect’s cache policies, with result persistence enabled because Prefect’s documentation says caching requires persisted results.

import asyncio, hashlib, os, tempfile, time
os.environ["PREFECT_HOME"] = tempfile.mkdtemp()            # fresh settings, database and result storage

from prefect import flow
from prefect.assets import Asset, add_asset_metadata, materialize
from prefect.cache_policies import INPUTS
from prefect.client.orchestration import get_client
from prefect.testing.utilities import prefect_test_harness

RAW = Asset(key="warehouse://sales/raw_orders")
DAILY = Asset(key="warehouse://sales/daily_sales")
SOURCE = {"2024-10-02": [120, 80, 45, 60]}                 # toy source rows
executions = []

def revision(day):
    return hashlib.sha256(repr(SOURCE[day]).encode()).hexdigest()[:8]

@materialize(RAW)
def load_raw(day: str) -> None:
    pass                                                   # stands in for copying the day's rows

@materialize(DAILY, asset_deps=[RAW], cache_policy=INPUTS, persist_result=True)
def build_by_day(day: str, upstream: None) -> int:         # cache key: the day only
    executions.append("by_day")
    add_asset_metadata(DAILY, {"partition": day, "input_revision": revision(day)})
    return sum(SOURCE[day])

@materialize(DAILY, asset_deps=[RAW], cache_policy=INPUTS, persist_result=True)
def build_by_revision(day: str, input_revision: str, upstream: None) -> int:   # cache key: day and revision
    executions.append("by_revision")
    add_asset_metadata(DAILY, {"partition": day, "input_revision": input_revision})
    return sum(SOURCE[day])

@flow
def pipeline_by_day(day: str) -> int:
    return build_by_day(day, load_raw(day))

@flow
def pipeline_by_revision(day: str) -> int:
    return build_by_revision(day, revision(day), load_raw(day))

A short helper, not shown, reads events whose names start with prefect.asset from the server’s events endpoint and prints the new ones after each run. The runs use Prefect’s temporary test server and a fresh PREFECT_HOME so earlier results cannot be reused.

with prefect_test_harness():
    print("run 1:", pipeline_by_day("2024-10-02"), "| builds executed:", len(executions)); new_events()
    print("run 2, same inputs:", pipeline_by_day("2024-10-02"), "| builds executed:", len(executions)); new_events()
    SOURCE["2024-10-02"][2] = 95                           # the source is corrected: 45 -> 95
    print("run 3, key is the day, source corrected:", pipeline_by_day("2024-10-02"), "| builds executed:", len(executions)); new_events()
    print("run 4, key includes the revision:", pipeline_by_revision("2024-10-02"), "| builds executed:", len(executions)); new_events()

run 1: 305 | builds executed: 1
    materialization.succeeded raw_orders   upstream=[] payload={}
    referenced               raw_orders   upstream=[] payload={}
    materialization.succeeded daily_sales  upstream=['raw_orders'] payload={'partition': '2024-10-02', 'input_revision': '035a9e3e'}
run 2, same inputs: 305 | builds executed: 1
    materialization.succeeded raw_orders   upstream=[] payload={}
run 3, key is the day, source corrected: 305 | builds executed: 1
    materialization.succeeded raw_orders   upstream=[] payload={}
run 4, key includes the revision: 355 | builds executed: 2
    materialization.succeeded raw_orders   upstream=[] payload={}
    referenced               raw_orders   upstream=[] payload={}
    materialization.succeeded daily_sales  upstream=['raw_orders'] payload={'partition': '2024-10-02', 'input_revision': 'ca41de09'}

Run 1 shows the event stream. raw_orders is materialized and also referenced because build_by_day consumed it. daily_sales is materialized, its event names raw_orders as upstream, and its payload carries the metadata added with add_asset_metadata, including the input revision 035a9e3e.

Run 2 passes the same inputs. The build did not execute (the counter stays at 1) and no daily_sales event appears; Prefect’s documentation says cached executions emit no asset events. That is accurate, since nothing was produced, but the asset’s history now holds no record that a run relied on the cached version.

Run 3 is the failure. The source was corrected after run 1, the cache key is still just the day, and the flow returned the old 305 from the cache. Nothing in the asset’s event history shows the problem: no event was emitted, and the last materialization still names revision 035a9e3e. This is the partition-key-as-snapshot-identifier mistake from the Dagster section, now appearing in a cache key. Run 4 puts the revision in the inputs, so the key changes, the task runs, and the new event records revision ca41de09 with the corrected total of 355. Dagster’s data-version tags solve the same problem by making input versions part of the record; in Prefect the author must put the version into the cache key and the metadata.

Prefect’s default cache policy, according to its documentation, combines the task’s inputs, its code definition, and the flow run ID, so by default results are not reused across runs. The behavior above came from choosing INPUTS, which is a common choice for expensive tasks. It is safe only when every input that determines the result is an argument.

Two other differences are worth planning for. The asset API identifies an asset by its key, and the examples carry the day as metadata; a design that needs per-day status in the asset view would have to encode the day in the key or track it elsewhere. And checks are ordinary control flow. A validation task that raises stops the flow body before later tasks execute, and the flow run ends in a failed state because an exception fails the run, according to the flows documentation. The placement rule from the previous section applies unchanged: the validation belongs in the task that publishes, or in a task whose result the publishing task consumes, so that no way of launching the publication skips it.

Choosing an orchestration model

The asset model earns its cost when the pipeline produces durable named objects that other people read, and when the questions that matter are about those objects: what is stale, what must be rebuilt after a correction, which version fed a report. The partition mapping example is such a question. A task graph can answer it too, but the author has to encode the window logic in the schedule and in the clear or backfill commands, DAG by DAG.

The task model fits better when the workflow is mostly actions: deploy a service, call a vendor, send a notification, orchestrate systems whose outputs are not durable data objects. Writing an asset for every side effect adds declarations without adding questions an operator would ask. A platform can keep both, with asset-oriented definitions for transformations and tasks for the glue, which makes the boundary between them a design decision to write down.

The asset model also has costs. Each asset needs a stable key, a chosen granularity, and declared reads, and the plan is no better than the declarations, as the first program showed. A pipeline with 200 assets and 365 daily partitions each (hypothetical numbers) produces 73,000 partition records per year before any rebuild, so the event log needs a retention and indexing plan. If the transformation layer is dbt, there are already two graphs, the SQL graph and the orchestrator’s, and keeping them consistent is ongoing work.

Between the two tools in this article the difference is where structure lives. Dagster declares the graph, partitions, versions, and checks up front, and in return it can answer questions before running anything, such as the reverse traversal above. Dagster can also launch assets from their status: its declarative automation documentation describes conditions such as eager, which updates an asset when its dependencies change, on_cron, and on_missing, evaluated by an automation condition sensor. Prefect keeps the graph dynamic and Pythonic, which suits workflows whose shape depends on the data, and it adds asset events and lineage on top. The cost is that more of the version and partition discipline is the author’s. Neither tool decides whether the data is right; they record who produced what and when, and they enforce the gates that you place.

What goes wrong beyond the examples

Asset history attaches to the key. If an asset is renamed, split, or merged, its record starts empty unless the history is migrated, and a report that cites the old key finds nothing. Keeping keys stable, and versioning the physical objects behind them as in the second program, avoids most of this.

The record store is state. If the orchestrator’s database is restored from an older backup, “last materialized” statements revert while the destination keeps the newer tables, so assets look stale or missing when they are not. After a restore, compare the records with the destination before acting on staleness; Airflow Operations walks through that scenario for a task-based scheduler, and the same discipline applies here.

Two runs can build the same partition at once, for example a scheduled rebuild and a manual backfill. The one that finishes last wins, even if it read older input. A publication step that compares the candidate’s input revision with the live version and refuses to move backward prevents this; the idea is the same as a fencing token.

Finally, the graph is a map of declared reads. Metadata and Lineage for Change Impact Analysis examines how to measure how much of the real graph a declared one covers, which is the question the finance_export example raised.

References

• Dagster documentation: Assets
• Dagster documentation: Testing assets with asset checks
• Dagster documentation: Defining dependencies between partitioned assets
• Dagster documentation: How backfills work
• Dagster documentation: Asset versioning and caching
• Dagster documentation: Declarative Automation
• Prefect 3 documentation: Assets
• Prefect 3 documentation: Flows
• Prefect 3 documentation: Caching
• Apache Airflow documentation: Asset-aware scheduling