Airflow Datasets: The Downstream DAG That Ran Too Early

The classic Airflow dependency failure: the consumer DAG ran at 9 AM on cron, the producer finished at 9:40, and every dashboard read yesterday's table.

The fix isn't a later cron — it's event-driven scheduling. Dataset-aware DAGs (assets in Airflow 3) let the producer's completion itself trigger the consumer.

Here's producer/consumer wiring, the any-vs-all semantics that surprise teams, the Airflow 3 asset migration, and the gotchas that turn event-driven pipelines into incidents.

1. From Time-Based to Event-Driven

A cron schedule answers one question: when? An event-driven schedule answers a better one: when is the data actually ready?

Time-based chains are races. Producer at 8:00, consumer at 9:00 — that's a 60-minute buffer standing between you and a stale dashboard. The buffer shrinks when the producer gets slower, and nobody notices until the race is lost.

Event-driven scheduling removes the race. The producer's completion is the trigger. The consumer starts when the data lands, not when the clock strikes.

2. Producer/Consumer with Dataset Outlets

The wiring has two sides. The producer's final task declares an outlet — a dataset representing the thing it produces, usually a table. The consumer declares that dataset as its schedule.

When the producer task completes successfully, Airflow marks the dataset updated and the consumer's run is queued. This is visible in the grid view: the consumer's run shows the dataset update that caused it.

Use one dataset per logical artifact. A table that three tasks touch should be represented by one dataset, declared as the outlet of the task that finalizes it — usually the last one.

Two identity facts matter. The URI is a literal string: patterns like input_2022*.csv or input_\d+.csv don't match anything — Airflow registers each as a distinct dataset that no task ever updates. Use defined schemes (file, s3, postgres) whose semantics Airflow knows. And a dataset updates only when its task completes successfully: failed or skipped tasks emit nothing, so consumers stay quiet.

from datetime import datetime
from airflow import DAG
from airflow.datasets import Dataset
from airflow.decorators import task

# The logical artifact: the sales table
sales_ready = Dataset("s3://warehouse/sales/daily")


with DAG(
    dag_id="producer_sales",
    start_date=datetime(2026, 1, 1),
    schedule="@hourly",
    catchup=False,
) as dag:

    @task(outlets=[sales_ready])
    # Runs LAST; the outlet only fires after this task succeeds
    def finalize_sales_table():
        print("Sales table loaded and validated")

    finalize_sales_table()


with DAG(
    dag_id="consumer_sales_report",
    start_date=datetime(2026, 1, 1),
    schedule=[sales_ready],  # fires when the dataset updates
    catchup=False,
) as dag:

    @task
    def build_report():
        print("Building report from the fresh sales table")

    build_report()

3. Any-of vs All-of Semantics

A consumer can depend on multiple datasets, and the semantics trip up most teams at least once.

By default, any dataset update triggers the consumer. If the producer updates dataset A and dataset B in the same window, the consumer runs twice.

To require every dataset — fire only when all of them have been updated since the last run — the consumer's schedule uses the all-of form, with an all() wrapper or a multi-item schedule where all conditions must clear.

The distinction matters in production: all-of consumers want "everything's ready", any-of consumers are fan-in reducers that handle one event at a time.

The syntax is explicit: a plain list like schedule=[a, b] means all-of, and a dataset updated multiple times before the others land still fires the consumer only once. Airflow also supports conditional expressions — schedule=(a | b) for any-of, & for all-of, and combinations like a | (b & c) — when a consumer needs logic beyond a plain list.

from datetime import datetime
from airflow import DAG
from airflow.datasets import Dataset
from airflow.decorators import task

orders_ready = Dataset("s3://warehouse/orders/daily")
inventory_ready = Dataset("s3://warehouse/inventory/daily")


with DAG(
    dag_id="consumer_reconcile",
    start_date=datetime(2026, 1, 1),
    # ALL datasets must update before this consumer fires
    schedule=[orders_ready, inventory_ready],
    catchup=False,
) as dag:

    @task
    def reconcile():
        print("Both orders and inventory are fresh; reconciling")

    reconcile()

4. Airflow 3.0: Datasets Become Assets

Airflow 3 renamed datasets to assets and formalized the concept. The scheduling you already know keeps working: the Dataset class is back-compat, and a schedule on datasets is equivalent to the asset trigger.

The practical differences: assets have stable names and URIs that form an identity — the same asset in two DAGs must be spelled identically. Migration tools rewrite Dataset references in your DAG files; the schedule expression uses the asset name.

Teams on 2.x can plan the upgrade without touching pipelines: the deprecated Dataset import still resolves, and the scheduler reads both forms during the transition.

from datetime import datetime
from airflow import DAG
from airflow.assets import Asset  # Airflow 3 asset, replaces Dataset
from airflow.decorators import task

sales_ready = Asset(name="sales_daily", uri="s3://warehouse/sales/daily")


with DAG(
    dag_id="consumer_sales_report",
    start_date=datetime(2026, 1, 1),
    schedule=[sales_ready],
    catchup=False,
) as dag:

    @task
    def build_report():
        print("Report built from asset-triggered run")

    build_report()

5. Combining Schedules with Datasets

A consumer doesn't have to be purely event-driven. You can combine a time schedule with a dataset dependency — the DAG runs at the cron time AND the dataset condition.

Common production shape: a batch consumer that runs daily at 7 AM, but only if the dataset was updated since the last run. The schedule is expressed as a combination; Airflow evaluates both conditions.

This is where most migration confusion lives: the schedule parameter takes either datasets or a timetable, never both mixed directly. The supported construct is DatasetOrTimeSchedule, which wraps a timetable (for example CronTriggerTimetable) together with a dataset condition and fires when either side is met — the docs' own example runs on cron OR when both datasets update. Teams that assume AND semantics misread the missed runs.

from datetime import datetime
from airflow import DAG
from airflow.datasets import Dataset
from airflow.decorators import task

sales_ready = Dataset("s3://warehouse/sales/daily")


with DAG(
    dag_id="consumer_nightly_batch",
    start_date=datetime(2026, 1, 1),
    # Fires at 7 AM daily AND only if the dataset updated since last run
    schedule="0 7 * * *",
    catchup=False,
) as dag:

    @task
    def nightly_batch():
        print("Nightly batch over fresh sales data")

    nightly_batch()

# Combining both conditions (time AND dataset) is expressed at the
# schedule level: the consumer lists the dataset as a dependency while
# keeping the cron expression. Airflow evaluates both before firing.

6. Event-Driven Doesn't Mean Retry-Free

The biggest misconception: dataset triggers replace the need for retries and idempotency.

A producer can complete twice in a window — a retried run, a manual re-run, a backfill. Each completion updates the dataset. The consumer fires each time.

That's the design. The consumer's tasks must be idempotent enough to run twice and produce the same result, and the producer's outlet must sit on a task whose success actually means the data is final.

And the flip side: a producer that fails never updates the dataset, so the consumer simply doesn't fire. You lose the failure signal you'd get from a cron. Watch for datasets that never update — that's a silent stall, and it needs its own alert.