Assets: Scheduling on Data, Not the Clock

Airflow 3's Assets let a DAG run the moment fresh data lands, instead of guessing the upstream lag with cron.

Every cron schedule is a bet. You write schedule="0 3 * * *" on the transform DAG and you’re wagering that the ingest job upstream will always be done by 3am. Most days it is. Then one morning the extract runs long, the transform fires against yesterday’s data, and you spend the standup explaining a number that’s off by a day. The clock didn’t lie — it just doesn’t know anything about your data.

Airflow 3 gives you a way out of the bet. Instead of scheduling on time, you schedule on data: a task declares that it produces a piece of data, and a downstream DAG says “run me whenever that data updates.” The scheduler wires the two together. No cron, no sensor polling, no guessing the lag.

Assets, formerly Datasets

If you read a 2.x tutorial, you saw this feature called a Dataset. It’s the same idea, renamed in 3.x to Asset — a logical handle to a piece of data, identified by a URI. The URI is just a name the scheduler tracks; it doesn’t have to point at anything Airflow can open.

from airflow.sdk import Asset

raw_orders = Asset("s3://bookshop/raw/orders")

A task becomes a producer by listing the asset in its outlets. When that task finishes successfully, Airflow records an update to the asset — that’s the event everything downstream keys off.

from airflow.sdk import dag, task

@dag(schedule="@daily", catchup=False)
def ingest_orders():
    @task(outlets=[raw_orders])
    def extract_and_load():
        # pull yesterday's orders from the store API, write to S3
        ...

    extract_and_load()

ingest_orders()

The ingest DAG still runs on a clock — something has to kick off the daily pull. But that’s the only clock in the chain. Everything downstream rides the data.

Scheduling on the asset

The transform DAG drops cron entirely. Its schedule is a list of assets, and it runs when they update.

@dag(schedule=[raw_orders], catchup=False)
def transform_orders():
    @task
    def build_marts():
        # read raw_orders, write the daily sales mart
        ...

    build_marts()

transform_orders()

The moment extract_and_load succeeds and marks raw_orders updated, the scheduler queues a run of transform_orders. If ingest is done at 2:58, the transform starts at 2:58. If ingest limps in at 4:15 after a retry, the transform starts at 4:15 — against the right data, every time. You’ve replaced a fragile assumption about timing with a fact about lineage.

This is cross-DAG choreography with no sensor in sight. The old pattern — an ExternalTaskSensor in the downstream DAG poking the upstream one until it went green — meant a worker slot held hostage, poll intervals to tune, and timeouts to get wrong. Assets invert it: the producer announces, the scheduler routes, nobody polls.

URIs, names, and why identity matters

Before you scale this past two DAGs, understand what actually identifies an asset — because the scheduler matches producers to consumers by identity, and getting the identity wrong means a trigger that silently never fires.

An asset has two identifying fields: a name and a uri. In 3.x you can construct one with either or both. If you pass a single positional string that looks like a URI, it becomes the uri and the name defaults to it; pass name= explicitly and you get a named asset whose URI may differ.

from airflow.sdk import Asset

# URI is the identity; name defaults to the same string
a = Asset("s3://bookshop/raw/orders")

# name is the identity, uri is metadata pointing at the physical store
b = Asset(name="raw_orders", uri="s3://bookshop-prod/raw/orders/")

The scheduler treats the URI as a canonical key, and it normalizes it. The URI must be a valid URI with a scheme, it must not contain a fragment (#...), and Airflow parses known schemes to canonicalize them — a postgres:// URI gets its host lowercased and default port folded in, for instance. Two producers that write s3://Bucket/x and s3://bucket/x are not guaranteed to resolve to the same asset once normalization runs; two that disagree on a trailing slash may or may not, depending on scheme. The failure mode is quiet: the consumer waits on s3://bucket/orders, the producer emits s3://bucket/orders/, and the trigger never fires because, as far as the scheduler is concerned, those are different assets. The fix is discipline — define each asset once, in a shared module, and import it into both producer and consumer:

# bookshop/assets.py — the single source of truth
from airflow.sdk import Asset

RAW_ORDERS = Asset("s3://bookshop/raw/orders")
RAW_INVENTORY = Asset("s3://bookshop/raw/inventory")
DAILY_SALES = Asset("s3://bookshop/marts/daily_sales")

Both DAGs from bookshop.assets import RAW_ORDERS. Now the identity is literally the same Python object, and a typo becomes an ImportError at parse time instead of a trigger that quietly does nothing at 3am.

Asset expressions: AND, OR, and the messy real cases

A mart usually needs more than one source. Pass several assets in the list and, by default, the DAG runs when all of them have updated since its last run — an implicit AND across your inputs:

@dag(schedule=[RAW_ORDERS, RAW_INVENTORY], catchup=False)
def daily_report():
    ...

But real dependencies aren’t always a flat AND. Maybe the report should run when orders land or when a manual correction file drops. Maybe it needs orders and either of two regional inventory feeds. For that, 3.x gives you AssetAny and AssetAll, and — more ergonomically — operator overloading on the assets themselves. & builds an AND (AssetAll), | builds an OR (AssetAny), and they nest:

from airflow.sdk import Asset, AssetAny, AssetAll

RAW_ORDERS = Asset("s3://bookshop/raw/orders")
INV_US = Asset("s3://bookshop/raw/inventory/us")
INV_EU = Asset("s3://bookshop/raw/inventory/eu")
CORRECTION = Asset("s3://bookshop/raw/corrections")

# orders AND (either inventory feed), OR a manual correction drop
condition = (RAW_ORDERS & (INV_US | INV_EU)) | CORRECTION

@dag(schedule=condition, catchup=False)
def daily_report():
    @task
    def build():
        ...
    build()

daily_report()

That expression reads exactly like the sentence you’d say in the standup: “run when orders and one of the inventory feeds have landed, or whenever a correction comes in.” The overloaded form is sugar for the explicit constructors, which you can still reach for when you’re building a condition programmatically:

condition = AssetAny(
    AssetAll(RAW_ORDERS, AssetAny(INV_US, INV_EU)),
    CORRECTION,
)

One subtlety worth internalizing: an AssetAll fires when each of its members has updated at least once since the DAG’s last asset-triggered run. It does not require them to update simultaneously, or within a window. If orders update at 01:00 and inventory at 05:00, the AND is satisfied at 05:00 and the run fires then, carrying both updates. If orders then update three more times before inventory moves again, those extra updates are folded into the next satisfaction — the scheduler tracks “has each leg seen an update since we last ran,” not a count. That’s usually what you want, but it means you can’t assume one downstream run per upstream update. Design the consuming task to read the current state of its inputs, not to assume it’s processing exactly one delta.

Data and time together: AssetOrTimeSchedule

Sometimes you need both a data trigger and a time floor: “run when orders land, but also run at 6am no matter what, so the morning report exists even if ingest is down.” That’s AssetOrTimeSchedule — it marries a timetable with an asset condition, and fires on either.

from airflow.sdk import Asset
from airflow.timetables.assets import AssetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

@dag(
    schedule=AssetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"),
        assets=(RAW_ORDERS & RAW_INVENTORY),
    ),
    catchup=False,
)
def daily_report():
    ...

daily_report()

Now the report runs the instant both sources land — and if they never do, the 6am tick guarantees a run anyway (against whatever data exists). You’re not choosing between data-driven and time-driven; you’re composing them. This is the pattern for anything with an SLA on existence rather than freshness: financial close reports, compliance extracts, the dashboard someone opens at 8am and expects to be populated. The asset condition gives you freshness on the good days; the timetable gives you a floor on the bad ones.

The @asset decorator and producing chains

For a DAG whose whole purpose is to produce one asset, the @asset decorator collapses the boilerplate. The function is the producing task, and the asset takes the function’s name.

from airflow.sdk import asset

@asset(schedule="@daily")
def raw_orders():
    # returns / writes the day's orders
    ...

This defines the asset and the DAG that maintains it in one move. What makes it more than sugar is that @asset-decorated functions compose into chains. A downstream asset can name an upstream one as a parameter, and Airflow injects the upstream asset’s event context and schedules the downstream on it:

from airflow.sdk import asset

@asset(schedule="@daily")
def raw_orders():
    # write the day's raw orders
    ...

@asset(schedule=raw_orders)
def clean_orders(raw_orders):
    # `raw_orders` is the upstream AssetEvent context;
    # this asset is produced whenever raw_orders updates
    ...

When one function genuinely produces several assets, @asset.multi lets a single task update more than one outlet at once — useful when a job writes a fact table and its late-arriving-dimension side table in the same transaction and you want both signals emitted together:

from airflow.sdk import asset, Asset

@asset.multi(
    schedule=raw_orders,
    outlets=[Asset("s3://bookshop/marts/sales"), Asset("s3://bookshop/marts/returns")],
)
def split_marts(raw_orders):
    # write both marts, emit both asset updates on success
    ...

Reach for the decorator when the DAG and the asset are one-to-one (or the multi case); drop back to explicit outlets on @task when a task updates assets as a side effect of doing something larger.

Metadata: attaching payloads to an update

An asset update doesn’t have to be a bare “something changed” ping. You can attach an extra payload — a JSON-serializable dict — that rides along with the event and is readable by every downstream consumer. This is how you pass what changed, not just that it changed: a row count, an S3 key, a watermark, a data-quality verdict.

There are two ways to emit metadata. The declarative way, when you know the payload at return time, is to yield a Metadata object from the task:

from airflow.sdk import Asset, Metadata, task

RAW_ORDERS = Asset("s3://bookshop/raw/orders")

@task(outlets=[RAW_ORDERS])
def extract_and_load():
    key = write_orders_to_s3()          # returns e.g. "raw/orders/2026-05-10.parquet"
    n = count_rows(key)
    yield Metadata(RAW_ORDERS, extra={"s3_key": key, "row_count": n})

The imperative way, when you’re deep in logic and want to attach metadata (or emit the update conditionally), is outlet_events from the task context:

from airflow.sdk import Asset, get_current_context, task

RAW_ORDERS = Asset("s3://bookshop/raw/orders")

@task(outlets=[RAW_ORDERS])
def extract_and_load():
    key = write_orders_to_s3()
    n = count_rows(key)
    context = get_current_context()
    context["outlet_events"][RAW_ORDERS].extra = {"s3_key": key, "row_count": n}
    # you can also skip the emit entirely by not touching the outlet on a no-op run

Either way, the payload is now part of the asset event, versioned in the metadata DB alongside the timestamp.

Reading events downstream: inlet_events and triggering_asset_events

The consumer side is where metadata earns its keep. A downstream task can read the events that triggered its run — so instead of re-scanning S3 to find what’s new, it reads the S3 key straight out of the event the producer emitted.

Two accessors matter. triggering_asset_events gives you exactly the events that caused this run to be scheduled — the OR/AND legs that fired. inlet_events[asset] gives you the full history of events for an asset you’ve declared as an inlet, most-recent last, which is how you reach back to “the previous watermark” for an incremental load.

from airflow.sdk import Asset, dag, get_current_context, task

RAW_ORDERS = Asset("s3://bookshop/raw/orders")

@dag(schedule=[RAW_ORDERS], catchup=False)
def transform_orders():
    @task(inlets=[RAW_ORDERS])
    def build_marts():
        context = get_current_context()

        # what fired *this* run
        for asset, events in context["triggering_asset_events"].items():
            for e in events:
                key = e.extra.get("s3_key")
                rows = e.extra.get("row_count")
                load_parquet(key, expected_rows=rows)

        # reach back to the prior event for an incremental watermark
        history = context["inlet_events"][RAW_ORDERS]
        if len(history) >= 2:
            last_key = history[-2].extra.get("s3_key")
            # process only what's newer than last_key
            ...

    build_marts()

transform_orders()

Note the two roles are separate: outlets/inlets declare the lineage edges (and grant access to the event accessors), while schedule= declares what triggers the run. A task can be an inlet reader without the asset being in the schedule, and vice versa. Declaring RAW_ORDERS as both the schedule trigger and an inlet — as above — is the common shape when you want the run to fire on the asset and read its payload.

AssetAlias: emitting to an asset you don’t know at parse time

Everything above assumes you know the asset’s identity when you write the DAG. Sometimes you don’t. A task partitions its output by date, or region, or tenant, and which asset it produces is decided at runtime — you can’t enumerate s3://bookshop/raw/orders/2026-05-10, .../2026-05-11, and so on in code that’s parsed once and runs forever. That’s what AssetAlias is for: a stable, parse-time handle that a task resolves to a concrete asset at run time.

You declare the alias in outlets, then, inside the task, add a real asset to the alias’s event. Downstream DAGs schedule on the alias and get triggered whenever the task resolves it to any concrete asset.

from airflow.sdk import Asset, AssetAlias, get_current_context, task

orders_partition = AssetAlias("bookshop-orders-partition")

@task(outlets=[orders_partition])
def extract_partition(logical_date):
    key = write_partition_to_s3(logical_date)   # e.g. raw/orders/2026-05-10.parquet
    concrete = Asset(f"s3://bookshop/raw/orders/{logical_date:%Y-%m-%d}")

    events = get_current_context()["outlet_events"]
    events[orders_partition].add(concrete, extra={"s3_key": key})

The .add(asset, extra=...) call is the runtime emission: it registers the concrete asset (creating it if it’s never been seen), attaches the metadata, and associates it with the alias. A consumer written like this needs to name neither the date nor the partition — it schedules on the alias and rides whatever the producer resolves:

@dag(schedule=[AssetAlias("bookshop-orders-partition")], catchup=False)
def index_partitions():
    @task(inlets=[AssetAlias("bookshop-orders-partition")])
    def index():
        context = get_current_context()
        for asset, events in context["triggering_asset_events"].items():
            for e in events:
                update_search_index(e.extra["s3_key"])
    index()

index_partitions()

The trade you’re making: an alias adds a layer of indirection that the asset graph can only fill in after the producer has run at least once, so the lineage looks empty until the first resolution. Reach for it only when the set of concrete assets genuinely isn’t knowable at parse time — dynamic partitions, per-tenant outputs. When you can enumerate them, a plain Asset per output is easier to read and shows up in the graph immediately.

data_interval on asset-triggered runs, and idempotency

Here’s the gotcha that bites people migrating time-based logic to assets. On a cron run, data_interval_start/data_interval_end bracket a real time window, and you lean on them for idempotency — “process the rows in [start, end).” On an asset-triggered run there is no natural time window: the run was caused by an event, not a tick. Airflow does populate a data_interval, but it’s effectively a point derived from the triggering event’s timestamp, not the meaningful business window your query wants.

The consequence: do not key your incremental logic off data_interval on asset-triggered DAGs. If the same asset event gets reprocessed (a manual re-run, a clear-and-rerun during an incident), a data_interval-based WHERE clause won’t protect you, because the interval is tied to when the event happened, not to the slice of data it represents. Instead, carry the business watermark in the asset extra (the s3_key, the max updated_at, a batch id) and make the consuming task idempotent on that — a MERGE/upsert keyed on the batch id, or a delete-insert scoped to the key from the event. You already have the mechanism from the section above; the discipline is to trust the payload for identity and treat the run’s data_interval as metadata about when, not what.

External events: the AssetWatcher and message queues

Everything so far is Airflow-produces, Airflow-consumes. The other half of “event-driven” — and the genuinely new capability in 3.x — is scheduling a DAG on an event that originates outside Airflow entirely. A file lands in a bucket and an SQS message announces it; a Kafka topic gets a record; an upstream system you don’t control pushes a notification. You want a DAG run then, without a sensor burning a worker slot polling.

That’s the AssetWatcher. You attach one or more watchers to an asset; each watcher wraps a trigger (the same deferrable-trigger machinery from the async world) that listens on a message queue. When a message arrives, the triggerer marks the asset updated — and anything scheduled on that asset runs. The listening happens in the triggerer process, asynchronously, so a thousand idle queues cost you almost nothing.

from airflow.sdk import Asset, AssetWatcher
from airflow.providers.amazon.aws.triggers.sqs import SqsSensorTrigger

# a trigger that long-polls an SQS queue for new messages
orders_landed_trigger = SqsSensorTrigger(
    sqs_queue="https://sqs.us-east-1.amazonaws.com/123456789012/bookshop-orders",
    aws_conn_id="aws_default",
)

orders_arrived = Asset(
    "s3://bookshop/raw/orders",
    watchers=[AssetWatcher(name="sqs_orders", trigger=orders_landed_trigger)],
)

@dag(schedule=[orders_arrived], catchup=False)
def ingest_on_arrival():
    @task
    def load():
        # the file is already in S3; the SQS message told us it landed
        ...
    load()

ingest_on_arrival()

Now the DAG runs on the external event — the message on the queue — not on a clock and not on an Airflow-produced asset. This is the closest Airflow gets to a true event-driven system: a producer outside your orchestrator drops a message, and the run happens in seconds. Providers ship watcher-compatible triggers for the common brokers (SQS, and Kafka-family message queues); the pattern generalizes to anything you can express as a deferrable trigger. The message’s contents can be threaded into the asset event’s extra, so the downstream task reads the S3 key from the same event that woke it up.

Seeing it in the UI

The api-server’s UI carries an asset view that turns all this wiring into a graph you can read: each asset as a node, producers feeding in, consumers hanging off, and a history of update events per asset — extra payloads included. When the daily report didn’t run, you don’t reverse-engineer a cron expression — you open the asset graph, find the input that never got its green update, and you’re already looking at the DAG to blame. The dependency you declared in code is the dependency you debug in the UI.

When assets are the wrong tool

Assets are a coordination mechanism inside one Airflow deployment. That’s their boundary, and it’s worth stating plainly so you don’t reach for them where they’ll disappoint:

  • Cross-team, cross-deployment. If the producer lives in another team’s Airflow instance, in-process asset events don’t cross the wire. You need a real broker between you — which is exactly the AssetWatcher-on-a-queue pattern above, not a shared Asset object. Assets coordinate; they don’t federate.
  • Non-Airflow producers. A Fivetran sync or a Spark job on someone else’s cluster can’t add to your outlet_events. Either have it drop a message on a queue an AssetWatcher listens to, or accept a time-based schedule with a sensor as the seam. Don’t fake an asset update from a DAG that merely checks whether the external job ran — at that point you’ve reinvented the sensor with extra steps.
  • Fan-out where every consumer needs a different slice. Assets signal “this changed.” If ten downstreams each need a genuinely different, computed view of the change, the asset gets you the trigger but not the routing — you still write the per-consumer logic. That’s fine; just don’t expect the asset to carry it.

Migrating from 2.x Dataset

If you’re bringing a 2.x pipeline forward: the concept is unchanged, the imports moved, and the names got shorter. Dataset becomes Asset; from airflow.datasets import Dataset becomes from airflow.sdk import Asset. The DatasetAny/DatasetAll expression classes become AssetAny/AssetAll (the &/| overloads work the same). DatasetOrTimeSchedule becomes AssetOrTimeSchedule. The task-context accessors were renamed in lockstep — outlet_events, inlet_events, and triggering_asset_events (the last was triggering_dataset_events in 2.x). The URIs you already defined carry over unchanged, so existing lineage survives the rename — but re-verify any URI you typed twice, because the identity rules and normalization described above are what a working trigger actually depends on.

Final thoughts

Cron scheduling encodes what you hope is true about timing. Asset scheduling encodes what’s actually true about dependencies — this data comes from that job — and lets the scheduler act on it. Composed with expressions, metadata, and external watchers, it stops being a clever way to skip a sensor and becomes the backbone of a pipeline whose structure lives in one place: the producer announces with a payload, the condition decides, the consumer reads what changed, and an external message can start the whole thing. That’s not just fewer failed mornings. It’s a pipeline where the question “did this run against fresh inputs?” stops being something you check and starts being something the scheduler guarantees.

Next: Executors: Local, Celery, and Kubernetes

Comments