How is an Asset different from a Dataset?

They are the same feature under a new name. Datasets shipped in Airflow 2.4; Airflow 2.9 began the rename to Asset, and Airflow 3.0 completed it — `airflow.sdk.Asset` is the current import, `airflow.datasets.Dataset` is deprecated. The scheduling behavior is unchanged, only the name and import path moved.

Do I still need a sensor to wait for an upstream file or table?

Not for anything produced by another DAG in the same Airflow instance. A sensor polls something external on an interval; an Asset-triggered DAG is woken immediately when the producing task finishes, with zero polling and zero wasted worker slots. Sensors still earn their place for genuinely external triggers — an S3 upload from outside Airflow, a third-party webhook — where nothing inside Airflow ever calls `outlets=`.

Can a DAG be scheduled on more than one Asset?

Yes. `schedule=[asset_a, asset_b]` runs the DAG once every listed Asset has been updated at least once since the last run — the default is 'all of them,' not 'any of them.' `AssetOr(asset_a, asset_b)` and `AssetAnd(...)` express other combinations explicitly.

What happens if the producing task fails?

The Asset is not updated, so nothing downstream fires. This is the whole point: a data-driven schedule cannot trigger on data that was never actually produced, which is a real failure mode a cron-based schedule cannot see — a time-based DAG runs on schedule whether or not yesterday's data actually landed.

Assets: Scheduling on Data, Not on Time

A cron schedule runs whether or not yesterday’s data actually arrived. An Asset-triggered DAG runs when a specific upstream task actually finishes — the schedule is data, not time.

Defining an Asset

from airflow.sdk import Asset, dag, task

orders_extracted = Asset("s3://bookshop/orders/extracted")

An Asset is just a URI with a name — a logical pointer to “the thing this DAG produces,” not a live connection to S3. Airflow never touches the URI itself; it only tracks whether a task declared that it updated the thing the URI names.

A producer that updates the Asset

# producer_dag.py
from airflow.sdk import Asset, dag, task

orders_extracted = Asset("s3://bookshop/orders/extracted")


@dag(schedule=None, catchup=False)
def extract_orders():
    @task(outlets=[orders_extracted])
    def extract():
        print("extracted 10 orders")

    extract()


extract_orders()

outlets=[orders_extracted] is the entire mechanism — when extract finishes successfully, Airflow records an update to that Asset. schedule=None means this DAG only runs when triggered manually or by an upstream system; it produces the Asset, it does not consume one.

A consumer scheduled on the Asset

# consumer_dag.py
from airflow.sdk import Asset, dag, task

orders_extracted = Asset("s3://bookshop/orders/extracted")


@dag(schedule=[orders_extracted], catchup=False)
def transform_orders():
    @task
    def transform():
        print("transformed orders after the asset updated")

    transform()


transform_orders()

schedule=[orders_extracted] replaces a cron expression entirely. This DAG has no time- based trigger at all — the only thing that creates a run is the named Asset being updated.

Running it

python -m venv .venv && source .venv/bin/activate
export AIRFLOW_VERSION=3.0.3
export PYTHON_VERSION=3.11
pip install "apache-airflow==${AIRFLOW_VERSION}" \
  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

cp producer_dag.py consumer_dag.py "$AIRFLOW_HOME/dags/"
airflow standalone

Confirm Airflow sees the Asset before triggering anything:

airflow assets list

name                           | uri                            | group | extra
===============================+================================+=======+======
s3://bookshop/orders/extracted | s3://bookshop/orders/extracted | asset | {}

Trigger the producer:

airflow dags trigger extract_orders && sleep 15
airflow dags list-runs extract_orders

dag_id         | run_id                                            | state   | run_after                        | logical_date | start_date                       | end_date
===============+===================================================+=========+==================================+==============+==================================+=================================
extract_orders | manual__2026-09-22T18:35:41.077647+00:00_hpoEXoHQ | success | 2026-09-22T18:35:41.077647+00:00 |              | 2026-09-22T18:35:42.662660+00:00 | 2026-09-22T18:35:44.590276+00:00

cat "$AIRFLOW_HOME/logs/dag_id=extract_orders/run_id=manual__2026-09-22T18:35:41.077647+00:00_hpoEXoHQ/task_id=extract/attempt=1.log"

{"timestamp":"2026-09-22T18:35:44.092507Z","level":"info","event":"extracted 10 orders","chan":"stdout","logger":"task"}

Now check the consumer — nobody triggered it:

airflow dags list-runs transform_orders

dag_id           | run_id                                              | state   | run_after                        | logical_date | start_date                       | end_date
=================+=====================================================+=========+==================================+==============+==================================+=================================
transform_orders | asset_triggered__2026-09-22T18:35:44.135908+00:00_6 | success | 2026-09-22T18:35:44.135908+00:00 |              | 2026-09-22T18:35:45.475791+00:00 | 2026-09-22T18:35:46.657046+00:00
                 | 1BqOV7p                                             |         |                                  |              |                                  |

The run_id itself says what happened: asset_triggered__..., not manual__... or scheduled__.... Airflow’s own naming records the cause of the run. Its log confirms it ran the real task body, not a no-op:

{"timestamp":"2026-09-22T18:35:45.727704Z","level":"info","event":"transformed orders after the asset updated","chan":"stdout","logger":"task"}

Under a second passed between the producer’s task finishing (18:35:44.59) and the consumer’s run starting (18:35:45.47) — no cron tick, no sensor poke interval, just the scheduler reacting to the Asset update.

Practice

1. Add a second producer task with its own outlet and confirm the consumer still fires from either one.

@task(outlets=[orders_extracted])
def extract_from_api():
    print("extracted 10 orders from the API")

Add this as a second task in extract_orders (not chained to extract) and trigger the DAG again:

airflow dags trigger extract_orders && sleep 15
airflow dags list-runs transform_orders

dag_id           | run_id                                              | state   | run_after
=================+=====================================================+=========+==================================
transform_orders | asset_triggered__2026-09-22T18:43:11.339941+00:00_l | success | 2026-09-22T18:43:11.339941+00:00
                 | k9U9gAJ                                             |         |
transform_orders | asset_triggered__2026-09-22T18:42:00.283516+00:00_f | success | 2026-09-22T18:42:00.283516+00:00
                 | 4XPZWNV                                             |         |

Exactly one new row (18:43:11), not two — transform_orders fires once per DAG run that updates the Asset, regardless of how many tasks inside that run declared the outlets=.

2. Schedule a DAG on two Assets and trigger only one of them.

customers_extracted = Asset("s3://bookshop/customers/extracted")

@dag(schedule=[orders_extracted, customers_extracted], catchup=False)
def join_orders_and_customers():
    ...

airflow dags trigger extract_orders && sleep 15
airflow dags list-runs join_orders_and_customers

No data found

No new run — despite extract_orders updating one of the two Assets this DAG is scheduled on. The default schedule=[a, b] semantics require every listed Asset to update at least once since the last run before the consumer fires; updating one of two is not enough.

3. Make the producer task fail and confirm the Asset is not updated.

@task(outlets=[orders_extracted])
def extract():
    raise ValueError("source system unreachable")

airflow dags trigger extract_orders && sleep 15
airflow dags list-runs extract_orders

dag_id         | run_id                                            | state  | run_after                        | logical_date | start_date                       | end_date
===============+===================================================+========+==================================+==============+==================================+=================================
extract_orders | manual__2026-09-22T18:39:42.155736+00:00_Eu1mR9Lv | failed | 2026-09-22T18:39:42.155736+00:00 | | 2026-09-22T18:39:43.565492+00:00 | 2026-09-22T18:39:55.118284+00:00

airflow dags list-runs transform_orders

dag_id           | run_id                                              | state   | run_after                        | logical_date | start_date                       | end_date
=================+=====================================================+=========+==================================+==============+==================================+=================================
transform_orders | asset_triggered__2026-09-22T18:35:44.135908+00:00_6 | success | 2026-09-22T18:35:44.135908+00:00 | | 2026-09-22T18:35:45.475791+00:00 | 2026-09-22T18:35:46.657046+00:00
                 | 1BqOV7p                                             |         |                                  |  |                                  |

Still one row — the same asset_triggered__ run from before the producer started failing. A failed task never reaches the point where Airflow records the outlet update — this is the exact failure mode a cron-scheduled consumer would miss, since it would run on schedule regardless of whether the producer actually succeeded.

4. Find the Asset's URI from inside a downstream task using context["triggering_asset_events"].

@task
def transform(**context):
    events = context["triggering_asset_events"]
    for asset, asset_events in events.items():
        print(f"triggered by: {asset.uri}")

{"event":"triggered by: s3://bookshop/orders/extracted","chan":"stdout","logger":"task"}

The consumer task can inspect exactly which Asset update caused its own run — useful once a DAG is scheduled on more than one Asset and the task body needs to branch on which one actually fired.

Defining an Asset

A producer that updates the Asset

A consumer scheduled on the Asset

Running it

Practice

Frequently Asked Questions