Asset-Triggered DAGs

Concepts covered: Cross-DAG dependencies

A cron schedule is a guess about when data is ready. An hourly orders feed that lands anywhere between 10 and 40 minutes past the hour forces the consumer DAG to pick a time: run at :45 and sit idle for most of each hour, or run at :20 and miss the late files. Assets replace the guess with a signal: the producing task tells Airflow when it finished, and the consumer runs then.

An asset is a declared name for data

An Asset is a name and a URI standing for data that Airflow never opens. Declaring one changes nothing in storage; it gives 2 DAGs a shared word for the same table or file. The task that updates the data lists the asset in outlets, and a DAG that should run after each update puts the same asset in schedule, in place of a cron string or a timetable.

When land reaches success, Airflow writes an asset event: the asset, the instant, and the producing DAG, task and run. On its next loop the scheduler reads the new events, finds that orders_summary is scheduled on orders_raw, and creates a run of it, usually within seconds of the event. Nothing in this path reads a clock, and nothing inspects storage: the event means that a task declaring the outlet succeeded. A task that declares the outlet and writes nothing still fires it, and a task that writes the file and then fails fires nothing.

from datetime import datetime
from airflow.sdk import Asset, dag, task
raw = Asset("orders_raw")
START = datetime(2026, 11, 10)

@dag(schedule="@hourly",
     start_date=START, catchup=False)
def orders_landing():
    @task(outlets=[raw])
    def land(): ...
    land()

@dag(schedule=[raw],
     start_date=START, catchup=False)
def orders_summary():
    @task
    def summarize(): ...
    summarize()
orders_landing()
orders_summary()

Asset-triggered runs carry events instead of a window

A scheduled run names a window, and its tasks read that window. An asset-triggered run has no window to name, because it exists for an update, and an update has no start and end of its own. Its logical_date is None, its data_interval_start and data_interval_end are None, and its run id starts with asset_triggered__ followed by the instant Airflow created it. A task that reads data_interval_start to pick a partition gets None and must be written for that. What the run does carry is triggering_asset_events in the task context: for each asset in the schedule, the list of events that caused the run, each holding its timestamp, source_dag_id, source_run_id and source_task_id. A consumer that needs the producer's window follows source_run_id back to the producing run and reads the window from there.

Events wait until a run consumes them

Events are kept until a run consumes them. If orders_summary is paused while the landing DAG succeeds 3 times, 3 events wait in its queue; on unpause the scheduler creates 1 run, and triggering_asset_events in that run lists all 3. The same batching happens whenever several events arrive between 2 scheduler loops. A consumer written to process everything since its last run handles this without noticing, and one written as if each run carried exactly 1 event reads the first and leaves 2 updates unprocessed.

The Assets page in the UI draws the graph the declarations imply: each asset in the middle, the DAGs whose tasks list it in outlets on one side, the DAGs scheduled on it on the other, and the time of its last event. In the grid of orders_summary, each run carries the asset-triggered marker in place of the clock, and the run's details list the events it consumed.

A producer whose whole job is 1 asset can be written with the @asset decorator from airflow.sdk, which builds a DAG of 1 task around the function and declares the asset as its outlet in the same step.

About this lesson

This section is part of the Scheduling: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.