Suppose one DAG fetches daily event data and another calculates statistics from it. An Airflow asset event connects the two: when the producer task successfully finishes with the data-set asset declared as an output, Airflow records an event. A consumer DAG scheduled on that asset can then run without relying on the producer’s clock schedule. Several consumer DAGs can listen to the same asset. section 4.2 section 4.6
The connection has two sides:
- In the producer, identify the data set with an Asset, such as Asset("file:///data/events_01"), and pass it as outlets=[events_dataset] on the task that writes the data. The producer can retain its daily schedule. section 4.3
- In the consumer, reference the same asset in schedule=[events_dataset]. Airflow triggers the DAG when the asset is updated and makes the triggering events available in its context. section 4.4
Find the data slice to process
An asset-triggered consumer cannot assume that its own logical_date identifies the producer’s data slice: unlike a time-scheduled run, it has no date- or interval-related schedule parameters. Instead, inspect triggering_asset_events. Its keys are assets and its values are lists of events; an event’s source_dag_run.logical_date can identify the producing run for an input filename. Alternatively, the producer can put a date in the event’s extra field for the consumer to read. With multiple input assets, select events by asset rather than taking the first event in the collection. section 4.4 section 4.7 section 4.9
A successful task with an outlet emits an event even if the underlying data did not change. If an existing output file means there is nothing new to process, the producer can raise AirflowSkipException; the skipped task emits no event and does not trigger consumers. Check that assumption before using file existence as the test for freshness. section 4.6
For a consumer scheduled on two assets, Airflow waits until both have seen an event; their update order does not matter. Asset scheduling therefore coordinates data availability, not matching dates or scheduling intervals across producers. Keep time-based scheduling where a fixed cadence is needed, and use asset events for downstream handoffs. section 4.7 section 4.9
Listing 4.5 Defining the input_path/output_path parameters based on source DAG
calculate_stats = PythonOperator(
task_id="calculate_stats",
python_callable=_calculate_stats,
op_kwargs={
"input_path": "/data/events/{{
➥ (triggering_asset_events.values() | first |
➥ first).source_dag_run.logical_date }}.json", #1
"output_path": "/data/stats/{{
➥ (triggering_asset_events.values() | first |
➥ first).source_dag_run.logical_date }}.csv", #2
},
)
Shows consumer input and output paths derived from the triggering event’s source DAG run rather than the consumer’s own logical date. section 4.4
