Assets Replace Datasets

Airflow 2.4 129 's datasets let a task declare that it updated a URI and another DAG run on that update instead of a clock. Airflow 3 renamed them assets (airflow.sdk.Asset), gave them names, and added asset aliases, asset watchers (Event-Driven Scheduling), partitions (3.2) and the @asset decorator, which defines an asset and the one-task DAG that builds it. Finance wants a report after every publish of BookNest's sales mart; the publish task declares the asset as an outlet (Writing DAGs), and the finance DAG is scheduled on it:

booknest_finance.py: a DAG scheduled on an asset, not a clockPython

from airflow.sdk import Asset, dag, task
DAILY_SALES = Asset(name="booknest_daily_genre_sales",
                    uri="postgres://l2-pg:5432/booknest/mart/daily_genre_sales")
@dag(schedule=[DAILY_SALES], tags=["booknest"])
def booknest_finance():
    @task
    def report(triggering_asset_events=None):
        for asset, events in triggering_asset_events.items():
            for e in events:
                print(f"{asset.name} updated by {e.source_dag_id}/{e.source_run_id}")
    report()
booknest_finance()

After the 29 June run of booknest_daily, the database held an asset event and the finance DAG had run:

Output of 8

            name            | source_dag_id  | source_task_id | source_run_id  | timestamp
----------------------------+----------------+----------------+----------------+-----------
 booknest_daily_genre_sales | booknest_daily | publish        | day-2026-06-29 | 20:57:26
booknest_finance  asset_triggered__2026-10-01T20:57:25.753228+00:00_0FlFhqu9  success
booknest_daily_genre_sales updated by booknest_daily/day-2026-06-29

Assets do not inspect data: Airflow records only that a successful task said it updated one, so name them after the real table. A list [a, b] waits for both; a | b runs on either.

MENU