When several teams need the same API data, separate ingestion pipelines duplicate requests and can produce inconsistent results. A producer DAG prepares the shared data once; consumer DAGs handle downstream work. The remaining question is how consumers know the data is ready. section 4.1
An Airflow asset connects the workflows without requiring the producer to name its consumers. The producer writes the data set and emits an asset event; consumer DAGs scheduled on that asset run when it is updated. The producer can retain a daily schedule while consumers react to updates rather than guessed completion times. section 4.2 section 4.1
Daily event ingestion
Define `events_dataset = Asset("file:///data/events_01")` and give the daily `fetch_events` task `outlets=[events_dataset]`. The task writes the day's events; its asset event can trigger a statistics DAG whose schedule references the same asset. Check the Assets tab to inspect the producer–asset relationship and emitted events. section 4.3 section 4.4
Listing 4.2 Adding the asset as an output of the fetch_events task
fetch_events = PythonOperator(
task_id="fetch_events",
python_callable=_fetch_events,
op_kwargs={
"start_date": "{{ data_interval_start | ds }}",
"end_date": "{{ data_interval_end | ds }}",
"output_path": "/data/events/{{ data_interval_start |
➥ ds }}.json",
},
outlets=[events_dataset], #1
)
The operator listing shows where the asset is declared as the output of the task that fetches events. section 4.3
Keep the event aligned with a real data change: a producer can skip downstream updates when a run did not modify the asset. It can also attach extra information to an event for consumers. If a consumer needs data from two producers, it can schedule on both assets; by default, it runs after each has produced an event. Asset-aware scheduling does not itself express fixed-time requirements. section 4.9 section 4.7
