Airflow Datasets: Event-Driven Triggers That Actually Work

The downstream DAG ran at 6 AM sharp. The upstream table landed at 6:40. That 40-minute gap cost a team a failed deploy and a very awkward status meeting.

Clocks can't see data. They fire because it's 6 AM, not because the table exists. Every fixed cron bakes in a guess about how long upstream takes, and guesses rot.

Assets flip the model. The consumer waits for the producer's event instead of the clock. You'll see how that wiring works here.

From Time-Based to Event-Driven

Time-based scheduling answers when, not whether. A cron fires at 6 AM because the clock says so, with zero knowledge of upstream state. That works while upstream is fast and breaks the first week it slows down. You'll relearn this lesson every backfill season.

Event-driven scheduling answers whether. The producer task emits an event when its output actually exists, and the consumer runs because the data is ready. Late upstream days become quiet waits instead of failed runs and manual retries.

The shift is small in code and large in operations. One schedule line changes from a cron string to an asset reference, and an entire class of timing pages disappears.

Producer and Consumer With Outlets

The producer declares its output as an outlet on the task that writes it. The scheduler records an asset event when that task succeeds. The consumer lists the same asset in schedule, so each event creates exactly one consumer run.

Names are the contract. A producer writing s3://lake/sales/dt={{ ds }} while the consumer subscribes to s3://lake/sales produces events that never match. Use one qualified naming convention and lint it.

Keep the producer on its clock and the consumer on the event. That split gives you predictable production cadence with patient downstream waits.

from airflow.sdk import Asset, dag, task

sales_asset = Asset("s3://lake/sales/dt=2026-08-01")

@dag(dag_id="sales_producer", schedule="@daily", catchup=False, tags=["sales"])
def sales_producer():
    @task(outlets=[sales_asset])
    def build_partition(logical_date=None):
        ds = logical_date.strftime("%Y-%m-%d")
        # write s3://lake/sales/dt=<ds>/ then emit outlet event
        return f"built partition {ds}"
    build_partition()

sales_producer()

@dag(dag_id="sales_consumer", schedule=[sales_asset], catchup=False, tags=["sales"])
def sales_consumer():
    @task
    def aggregate():
        return "read s3://lake/sales/dt={{ ds }}/ and build rollups"
    aggregate()

sales_consumer()

Multiple Datasets: All vs Any Semantics

Multiple inputs need explicit semantics. All-semantics waits for every listed asset to update before firing, which is what joins need. Any-semantics fires when any single asset updates, which suits independent feeds like per-region alerts. Airflow 3.x spells this as expressions: schedule=(a & b) waits for both, schedule=(a | b) fires on either, and you can nest them like a | (b & c) for mixed gates.

The default multi-asset behavior waits for all, but don't rely on memory. Declare the expression explicitly around join inputs so the next reader sees intent. Reserve OR for cases where a partial input is genuinely actionable. Repeated updates to one asset before the others catch up still produce a single consumer run, not one per event.

Test semantics by delaying one producer in staging. If the consumer fires early, you've got OR-semantics where you needed AND, or a misspelled asset that never participates.

Airflow 3 Asset Naming and Aliases

Airflow 3.x renamed Datasets to Assets and moved imports to airflow.sdk. The old airflow.datasets Dataset class still parses as back-compat, and schedules on datasets behave like asset triggers. That's the migration runway: old DAGs keep firing while you rename.

Don't straddle the rename forever. Mixed imports across producer and consumer DAGs create events under one name that subscriptions under the other name never see. Pick airflow.sdk Asset everywhere and convert in one pass with sed plus ruff AIR30 checks.

Aliases decouple teams. Producers emit concrete URIs while consumers subscribe to an alias, so URI evolution doesn't require rewriting every downstream DAG.

# Airflow 2.4+ style (still works on 3.x via back-compat)
# from airflow.datasets import Dataset
# sales = Dataset("s3://lake/sales/dt=2026-08-01")

# Airflow 3.x style (preferred)
from airflow.sdk import Asset, AssetAlias, dag, task

sales = Asset("s3://lake/sales/dt=2026-08-01")
# Alias lets producers evolve URIs without rewriting every consumer
sales_alias = AssetAlias("sales_daily")

@dag(dag_id="sales_producer_v3", schedule="@daily", catchup=False)
def producer():
    @task(outlets=[sales])
    def build(logical_date=None):
        return "write partition, emit Asset event"
    build()

producer()

Combining Schedules With Datasets

Combining clocks with events on one DAG double-fires. A consumer with both @daily and an asset schedule creates one run from the clock and one from the event, and both write the same partition. That's a restatement bug wearing a scheduling costume. When you genuinely need both, use the AssetOrTimeSchedule (AssetTimetable) integration so the DAG runs on either trigger under one timetable instead of two competing schedules.

Keep the roles clean: producers own clocks, consumers own events. When consumers need freshness SLAs, add elapsed-time alerting on event arrival instead of a competing cron. Alerting watches; crons duplicate.

Backfills complicate the picture because replaying producer history emits a burst of events. Plan consumer capacity for that burst or stagger producer history so downstream drains steadily. Queued asset events accumulate while the consumer is paused, then resolve on unpause, so a long pause followed by unpause can fire a burst you should expect.

Grid-View Lineage and Debugging Triggers

Grid view is your lineage debugger. Open the consumer run, follow the asset edge back to the producer event, and confirm timestamps line up. Missing edges mean missing events, not slow scheduling. The Asset Events tab on the run details page lists exactly which source events fired that run, and the Assets tab shows every registered asset with its producing tasks and subscribing DAGs.

Consumer tasks can read what fired them through triggering_asset_events: a dict of Asset to its AssetEvent list. Jinja templates pull data-interval bounds like {{ (triggering_asset_events.values() | first | first).source_dag_run.data_interval_start }}, and Python tasks accept it as a parameter to branch on run_id or timestamp. Declare assets a task reads as inlets so lineage shows consumption as well as production.

External systems push events through the REST API queued-event endpoints (/assets/queuedEvent and per-DAG variants), which suits producers outside Airflow. Pull-based AssetWatcher classes with BaseEventTrigger-compatible triggers poll queues or storage instead. When wiring looks right but runs still don't appear, check scheduler logs for asset-trigger evaluation errors. Rename mismatches after upgrades log quietly while the UI looks healthy.