Assets & Cross-DAG Data-Aware Scheduling

Until now, one DAG started at a set time. This chapter links DAGs to each other. One DAG can start when another DAG has made new data.

You follow shop_orders_daily again. Its last task, build_daily_revenue, writes the table daily_revenue. The run about 2026-01-20 is dag_run 380.

Other teams want that table. The DAGs of other teams in this chapter are made up: finance_report_daily and marketing_mail. All their times and ids are made up too.

First, see the problem. The finance team starts its report at 00:30 every day, because the table is usually ready by then. Look at two nights:

The report did not wait for the data. It waited for the clock. Airflow can do better: the report DAG can wait for the data itself.

The feature has two names. Airflow 2.4 added Datasets, and Airflow 3 renamed them Assets.

The idea is the same. Look at what changed:

You answer three questions:

1. How does a task say it made new data, and how does another DAG react?
2. What if the other DAG needs two datasets, or a clock as well?
3. When do you use a dataset, and when a sensor or a trigger?

A dataset is a name for a piece of data. The name is a URI: a text address such as a web address. Airflow does not open it.

A task says which datasets it makes with the argument outlets. When that task succeeds, Airflow records a dataset event. An event is one note: this dataset changed now.

Look at the producer. Only build_daily_revenue has outlets, because it writes the table. The URI is made up:

from datetime import datetime
from airflow import DAG
from airflow.datasets import Dataset
from airflow.operators.python import PythonOperator

daily_revenue = Dataset("postgres://warehouse:5432/analytics/shop/daily_revenue")

with DAG("shop_orders_daily", schedule="@daily",
         start_date=datetime(2026, 1, 1), catchup=False):
    load = PythonOperator(task_id="load_orders", python_callable=load_orders)
    build = PythonOperator(
        task_id="build_daily_revenue",
        python_callable=build_daily_revenue,
        outlets=[daily_revenue],
    )
    load >> build

Airflow keeps its events in the metadata database, where it saves all its state. Query it after run 380:

SELECT d.uri, e.source_task_id, e.source_run_id, e.timestamp
FROM   dataset_event e
JOIN   dataset d ON d.id = e.dataset_id
WHERE  e.source_dag_id = 'shop_orders_daily';

The event is created when the task succeeds, not when the whole DAG run ends. If build_daily_revenue fails, there is no event.

A consumer is a DAG that uses data made by another DAG. To wait for a dataset, you put the dataset in schedule.

Look at the finance DAG. It has no time, only the dataset. The URI must be the same text as in the producer:

from datetime import datetime
from airflow import DAG
from airflow.datasets import Dataset
from airflow.operators.python import PythonOperator

daily_revenue = Dataset("postgres://warehouse:5432/analytics/shop/daily_revenue")

with DAG("finance_report_daily", schedule=[daily_revenue],
         start_date=datetime(2026, 1, 1), catchup=False):
    PythonOperator(task_id="send_finance_report",
                   python_callable=send_finance_report)

The scheduler is the part of Airflow that starts runs. When it sees a new event for daily_revenue, it creates a run of finance_report_daily. It does this soon after the event.

Follow the night of 2026-01-21:

1. 00:00: run 380 of shop_orders_daily starts.
2. 00:00:14: build_daily_revenue succeeds. The dataset event is saved.
3. Soon after 00:00:14: the scheduler creates run 410 of finance_report_daily.

Now look at the two runs in the table dag_run:

SELECT id, dag_id, run_type, state
FROM   dag_run
WHERE  id IN (380, 410)
ORDER  BY id;

On the slow night of 2026-01-23, the finance run would start soon after 00:45. The finance report now always waits for fresh data.

Adding a second consumer needs no change to the producer. The producer does not know who listens.

A DAG can list more than one dataset. Then Airflow waits until all of them have a new event. It is an AND.

Take the made-up marketing_mail DAG. It needs daily_revenue and campaign_costs. Another made-up DAG, ads_costs_daily, writes campaign_costs.

daily_revenue = Dataset("postgres://warehouse:5432/analytics/shop/daily_revenue")
campaign_costs = Dataset("postgres://warehouse:5432/analytics/marketing/campaign_costs")

with DAG("marketing_mail", schedule=[daily_revenue, campaign_costs],
         start_date=datetime(2026, 1, 1), catchup=False):
    PythonOperator(task_id="send_mail", python_callable=send_mail)

Now follow one night. On the same night, someone also re-runs build_daily_revenue by hand at 00:50. Look at what each DAG does:

Look at the marketing column. Airflow remembers only that daily_revenue changed at least once since the last run. It does not count the changes.

At 02:30 both datasets have changed, so run 412 is created. Then the memory is cleared, and the next night starts from zero.

Look at the finance column too. A re-run of the producer sends a new event, so every consumer runs again. Plan for that.

Airflow 2.9 added conditions. You join datasets with & for AND, and with | for OR. Check your Airflow version before you use them.

Look at three schedules. Each uses the two datasets from the last section:

A condition cannot see silence. If an upstream DAG makes nothing, no event comes, and your DAG waits forever.

The fix is a clock as well. DatasetOrTimeSchedule, added in Airflow 2.9, runs the DAG when the datasets are ready. It also runs the DAG every day at the clock time.

Look at marketing_mail with a 07:00 clock. The time is made up:

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

with DAG(
    "marketing_mail",
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"),
        datasets=(daily_revenue & campaign_costs),
    ),
    start_date=datetime(2026, 1, 1),
    catchup=False,
):
    PythonOperator(task_id="send_mail", python_callable=send_mail)

A normal daily run is about one day. That day is its data interval, as the chapter Scheduling, Data Intervals & the Logical Date Trap showed.

A run started by a dataset is not about a producer day. Airflow only knows that something changed.

In Airflow 2, its logical date is the time of the newest event that started it. In Airflow 3, such a run has no logical date.

So the consumer must not guess the day from the run date. Look at the last two rows of daily_revenue:

SELECT revenue_date, revenue
FROM   daily_revenue
ORDER  BY revenue_date DESC
LIMIT  1;

The latest row works when events arrive one by one. A re-run of an old day would break it.

For more care, the producer can attach a small note to its event, such as the report date. The consumer can read the list of events that started its run. Check your Airflow version for the exact calls.

Airflow compares URI text. It never opens the table, and it never counts rows. An event means a task said it made data.

A small typo makes a different dataset. Look at two URIs that differ by one letter:

Three more facts follow from this:

• A task that succeeds but writes zero rows still sends an event.
• Both DAGs must live in the same Airflow system, with one metadata database.
• A job outside Airflow can send an event through the REST API, a web address that programs call. This exists from Airflow 2.9.

Look at the third fact. A Spark job outside Airflow can report that it wrote the table. The URL and login are made up:

curl -X POST "https://airflow.example.com/api/v1/datasets/events" \
  -H "Content-Type: application/json" \
  -u "svc_spark:secret" \
  -d '{"dataset_uri": "postgres://warehouse:5432/analytics/shop/daily_revenue"}'

Airflow trusts the events. So check the data inside the producer task, before it ends. A failed check then stops the event.

A sensor is a task that waits for something. The chapter Concurrency, Pools, Priorities & Deferrable Operators teaches how waiting costs worker slots.

Two older tools also link DAGs. ExternalTaskSensor waits in the consumer. TriggerDagRunOperator pushes from the producer.

Compare the three on the finance case. Each row asks who must know about whom:

Look at the sensor risk. ExternalTaskSensor looks for a producer run with the same logical date.

If the two DAGs have different schedules, no such run exists. It waits until its timeout, unless you set execution_delta.

Now choose. Use a dataset when you own the consumer and just need fresh data. Use a trigger when the producer must send settings, called conf, or wait for the result.

Use a sensor when you cannot change the producer. For example, another team owns it, and it sends no events.

A consumer DAG waits for data, not for the clock. The producer says what it made, and the scheduler connects the two.

Check the answers you should know:

Try this alone. Two teams need your table, and a third team will join next month. Which of the three tools keeps your DAG unchanged?

The answer is a dataset. The new team only adds the URI to its own schedule.