Airflow Dataset-Aware Scheduling: Producer/Consumer Triggers Across DAGs

airflow datasets are the primitive that finally deletes the fragile cross-DAG scheduling glue every senior data engineer has spent a decade writing by hand — the timing-hacks in cron expressions, the polling loops in ExternalTaskSensor, the fire-and-hope TriggerDagRunOperator chains that break the moment an upstream DAG slips fifteen minutes. Airflow 2.4 introduced datasets as a first-class scheduling object: a producer DAG declares what it emits via outlets=[Dataset("s3://...")], a consumer DAG declares what it listens for via schedule=[Dataset("s3://...")], and the scheduler wires the trigger without a single cron line, a single sensor, or a single retry configuration on either side. The engineering trade-off lives in how you name the dataset URI, which staleness semantics you promise across teams, and when you combine dataset triggers with cron via DatasetOrTimeSchedule — not in whether you should be using datasets in the first place.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "walk me through what happens when a producer DAG completes but downstream consumers don't fire," or "explain the difference between airflow data-aware scheduling and TriggerDagRunOperator," or "how would you migrate a 40-DAG ExternalTaskSensor graph to dataset trigger semantics without stopping the world?" It walks through why producer consumer dag orchestration was so painful before 2.4, the four axes interviewers actually probe (airflow dataset uri semantics, outlets, airflow schedule dataset shape, staleness), the producer side with airflow dataset outlet declarations, the consumer side with schedule=[Dataset(...)] and the AND semantics across multiple datasets, the cross-dag dependency and airflow event-driven story in the Airflow UI datasets tab, and the Airflow 3.x additions — dataset filters, custom triggers on datasets, and a complete ExternalTaskSensor migration recipe. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse SQL transformations on the SQL practice library →, and tune throughput and freshness with the optimization practice library →.

---

On this page

• Why dataset scheduling replaced ExternalTaskSensor
• Producer DAG — dataset outlets
• Consumer DAG — schedule on dataset
• Cross-team lineage + dataset UI
• Airflow 3.x additions + production migration
• Cheat sheet — Airflow datasets recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why dataset scheduling replaced ExternalTaskSensor

Before 2.4, cross-DAG orchestration was a graveyard of ExternalTaskSensor, TriggerDagRunOperator, and hand-tuned cron offsets

The one-sentence invariant: before Airflow 2.4, if DAG-B needed to run after DAG-A finished, you had exactly three bad options — align cron schedules with a hand-tuned offset, poll DAG-A's task instance state with ExternalTaskSensor, or push a trigger from A to B with TriggerDagRunOperator — and every one of them breaks the moment a DAG slips, a task retries, or a team renames a DAG. The pain compounded across teams: DAG-A owned by team-ingest, DAG-B owned by team-finance, and a Slack thread every quarter arguing whether the cron offset should be 15 or 20 minutes now that ingest is slower on Fridays. Datasets (Airflow 2.4+, hardened through 2.7 and expanded in 3.x) collapse all three anti-patterns into one primitive: a producer declares what it produces, a consumer declares what it consumes, and the scheduler wires the trigger without either team writing a single line of scheduling glue.

The four "must-answer" axes interviewers probe.

• Dataset URI. The URI is the contract between producer and consumer — s3://raw/orders, snowflake://finance/facts, custom://mkt/events. It is a string identifier, not a physical path check; the scheduler treats the string as an opaque event key. Naming conventions matter more than the underlying storage.
• Outlets on the producer. outlets=[Dataset("s3://raw/orders")] is declared on the task that produces the dataset, not the DAG. When the task instance reaches the success state, the scheduler emits a dataset event; anything else (failure, skip, upstream_failed) emits no event. The success-only semantics is the invariant every senior answer starts with.
• Consumer schedule shape. dag = DAG(schedule=[Dataset(...)]) replaces schedule_interval="0 6 * * *". Multiple datasets in the list mean all must have received events since the last run (AND semantics, not OR). DatasetOrTimeSchedule combines dataset triggers with a cron backstop.
• Staleness and freshness. The scheduler wires the trigger on any successful producer run; the consumer doesn't check whether the underlying data actually changed. Freshness SLAs are the operational surface — production teams instrument dataset event age and alert when a dataset hasn't been updated inside its expected window.

Why the pre-2.4 world was a graveyard.

• Cron offset alignment. DAG-A runs at 05:00; DAG-B runs at 05:30 assuming A finishes in 20 minutes. When A slips to 40 minutes, B runs on stale inputs and produces wrong numbers. Every offset is a lie about wall-clock behaviour that reality invalidates monthly.
• ExternalTaskSensor polling. DAG-B's first task is a sensor that polls DAG-A's task instance state on a schedule. The sensor holds a worker slot for the duration of the wait, contributes to concurrency pressure, and requires the polling DAG to know the DAG-ID and task-ID and execution-date arithmetic of the source DAG — a hard coupling across team ownership.
• TriggerDagRunOperator push. DAG-A's last task calls TriggerDagRunOperator(trigger_dag_id="dag_b"). The producer now owns knowledge of every downstream consumer, and adding a new consumer means editing the producer DAG. It is a one-way coupling that scales exactly as badly as callback hell.
• The retry hall of mirrors. A retry on DAG-A's producer task under any of the pre-2.4 models re-fires the sensor / re-triggers B in ways that either duplicate the downstream run or silently swallow it, depending on the exact combination of depends_on_past, execution_timeout, and reschedule mode. The sharp edges are notorious.
• Cross-team ownership fog. With ExternalTaskSensor, DAG-B's team must know DAG-A's internal task layout. Renaming a task in DAG-A silently breaks the sensor. There is no discoverability, no explicit contract, and no way for team-B to declare "I need whatever DAG-A produces without caring about how."

What datasets replaced these anti-patterns with.

• URI as contract. The dataset URI is a stable public string that team-A commits to emit and team-B commits to consume. Renames become public API changes, not silent breakage. Team-B never touches team-A's task layout.
• Success-only event semantics. The scheduler emits a dataset event exactly once per successful producer task run. Retries that eventually succeed emit one event; runs that never succeed emit none. The mental model is clean.
• AND across multiple datasets. schedule=[Dataset(a), Dataset(b), Dataset(c)] triggers when all three have received an event since the last consumer run. Downstream data models that need multiple sources arriving become a one-line declaration.
• UI as marketplace. The Airflow Datasets tab shows every declared dataset, its producer DAG(s), its consumer DAG(s), and the recent event history. Discoverability is free; team-C can subscribe to a dataset team-A already produces without any coordination.
• 2026 reality. Airflow 3.x adds dataset filters (consume only when a payload predicate is true), custom triggers on datasets, and richer event metadata. The 2.4 primitive was the foothold; 3.x is the mature surface.

What interviewers listen for.

• Do you say "the URI is a contract, not a path" in the first sentence? — senior signal.
• Do you name the success-only event semantics without being prompted? — required answer.
• Do you push back on "just use ExternalTaskSensor" with the ownership + polling + retry-hall-of-mirrors argument? — senior signal.
• Do you describe AND semantics across multiple datasets correctly? — required answer.
• Do you mention the Datasets tab as a lineage catalog? — senior signal.

Worked example — from ExternalTaskSensor to dataset trigger

Detailed explanation. The textbook migration case. Team-ingest runs ingest_orders_dag daily at 05:00 UTC; team-analytics runs transform_orders_dag that must start once ingest finishes. Historically, transform_orders_dag opened with an ExternalTaskSensor polling the ingest DAG's last task. Walk an interviewer through the before/after: what the sensor DAG looked like, what its failure modes were, and what the datasets version looks like end-to-end.

• The pre-2.4 sensor. ExternalTaskSensor(external_dag_id="ingest_orders_dag", external_task_id="load_to_s3", execution_delta=timedelta(0)).
• The failure modes. Worker slot held for the wait; execution_delta assumptions that break on DST; polling load on the metastore; sensor timeout races with the source DAG's slip.
• The datasets rewrite. One line on the producer, one line on the consumer, zero sensors.

Question. Rewrite the ingest producer and the transform consumer using Airflow datasets. Show the full DAG code for both sides and explain what the scheduler does at each event.

Input.

Component	Before (pre-2.4)	After (datasets)
Producer DAG	ingest_orders_dag (cron 05:00)	ingest_orders_dag (cron 05:00) + outlets
Consumer DAG	transform_orders_dag (cron 05:30 + ExternalTaskSensor)	transform_orders_dag (schedule=[Dataset])
Coupling	Consumer knows producer's task-id	Consumer knows only URI
Wait mechanism	Sensor polling	Scheduler event

Code.

# producer_dag.py — ingest_orders_dag
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime

orders_raw = Dataset("s3://raw/orders")

with DAG(
    dag_id="ingest_orders_dag",
    start_date=datetime(2026, 1, 1),
    schedule="0 5 * * *",          # producer keeps its cron
    catchup=False,
    tags=["ingest", "orders"],
) as producer:

    fetch = PythonOperator(
        task_id="fetch_orders",
        python_callable=lambda: None,
    )

    clean = PythonOperator(
        task_id="clean_orders",
        python_callable=lambda: None,
    )

    load = PythonOperator(
        task_id="load_to_s3",
        python_callable=lambda: None,
        outlets=[orders_raw],       # <-- the one line that changes
    )

    fetch >> clean >> load


# consumer_dag.py — transform_orders_dag
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime

orders_raw = Dataset("s3://raw/orders")

with DAG(
    dag_id="transform_orders_dag",
    start_date=datetime(2026, 1, 1),
    schedule=[orders_raw],          # <-- replaces cron + ExternalTaskSensor entirely
    catchup=False,
    tags=["analytics", "orders"],
) as consumer:

    stage = PythonOperator(task_id="stage", python_callable=lambda: None)
    tx    = PythonOperator(task_id="transform", python_callable=lambda: None)
    load  = PythonOperator(task_id="load_dw", python_callable=lambda: None)

    stage >> tx >> load

Step-by-step explanation.

1. The producer DAG keeps its cron schedule (0 5 * * *) — the change is a single outlets=[orders_raw] argument on the task that actually writes the S3 prefix. Only the load task carries the outlet; the fetch and clean tasks do not, because the scheduler should only emit an event when the final write succeeds.
2. When load_to_s3 reaches the success state, the Airflow scheduler emits a dataset event keyed by the URI s3://raw/orders. The event is atomic and idempotent — one success = one event, regardless of how many retries the task went through to get there.
3. The consumer DAG replaces its cron + sensor pair with schedule=[orders_raw]. The scheduler, on receiving the dataset event, looks up every DAG whose schedule list contains this dataset URI, checks the "since last run" condition, and enqueues a new DAG run for each matching consumer.
4. The consumer's first task (stage) starts immediately — no sensor, no wait, no worker slot held for polling. The DAG run's logical date is the dataset event timestamp, so downstream data-lineage queries can join runs to source events by timestamp.
5. If the producer task fails and doesn't retry to success, no dataset event is emitted; the consumer simply doesn't run. There is no timeout arithmetic, no execution_delta fudge factor, and no cross-team knowledge of DAG-ID or task-ID structure. The URI is the entire contract.

Output.

Metric	Before (ExternalTaskSensor)	After (dataset)
Consumer worker slots held during wait	1 per run	0
Coupling between teams	producer task-id	URI only
Rename resilience	breaks on rename	breaks on URI change (public contract)
Retry semantics	sensor timeout race	one success = one event
DAG lines of scheduling glue	~15	1 outlet + 1 schedule

Rule of thumb. Every ExternalTaskSensor in a modern Airflow codebase is a migration candidate. The rewrite is one line on each side; the operational win is a per-DAG worker slot freed and a per-quarter Slack thread avoided. If you inherit an Airflow deployment with any surviving ExternalTaskSensor, mark the migration in your first-week plan.

Worked example — the retry hall of mirrors under sensors vs datasets

Detailed explanation. A subtler failure mode of the pre-2.4 world: retries. An ingest task fails on its first attempt, retries three minutes later, and succeeds. Under ExternalTaskSensor polling with a naive configuration, the sensor either sees the intermediate "failed" state and gives up, or sees the final "success" state and fires — depending on exactly when the sensor happens to poll relative to the retry. Under datasets, the semantics are unambiguous: one final success = one event, regardless of retry count.

• The retry timeline. Producer task attempt 1 fails at 05:07; attempt 2 (retry) succeeds at 05:10.
• The sensor race. Sensor with poke_interval=60, mode=poke polls at 05:06 (task still running), 05:07 (state="failed"), 05:08 (task in retry queue), 05:09 (task running), 05:10 (state="success"). Depending on state-machine version, a poll at 05:07 may see failed and short-circuit even though the task will retry.
• The dataset alternative. Zero polling. One event emitted at 05:10 when the retry succeeds. Consumer runs from that event.

Question. Write the producer and consumer under both models. For the sensor version, enumerate the states the sensor may see. For the dataset version, show the single event trace.

Input.

Component	Value
Producer task	load_to_s3 with retries=3, retry_delay=timedelta(minutes=3)
Attempt 1 outcome	fail at 05:07
Attempt 2 outcome	success at 05:10
Sensor poke_interval	60 seconds
Sensor mode	poke

Code.

# Sensor version — the fragile pre-2.4 pattern
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.python import PythonOperator

with DAG(dag_id="transform_sensor", schedule="30 5 * * *", ...) as sensor_dag:
    wait = ExternalTaskSensor(
        task_id="wait_for_ingest",
        external_dag_id="ingest_orders_dag",
        external_task_id="load_to_s3",
        execution_delta=timedelta(minutes=30),
        poke_interval=60,
        mode="poke",
        allowed_states=["success"],
        failed_states=["failed"],   # <-- the trap: fires on any failed state
    )
    do_work = PythonOperator(task_id="do_work", python_callable=lambda: None)
    wait >> do_work


# Dataset version — the clean pattern
orders_raw = Dataset("s3://raw/orders")

with DAG(dag_id="transform_dataset", schedule=[orders_raw], ...) as ds_dag:
    do_work = PythonOperator(task_id="do_work", python_callable=lambda: None)

Step-by-step explanation.

1. Under the sensor version, the failure-state race is real: at 05:07 the ingest task instance's state is briefly failed between attempts. If failed_states=["failed"] is set, the sensor short-circuits at that poll and marks itself failed, even though the ingest task will retry and eventually succeed. The workaround is dropping failed_states=["failed"], but then a genuinely-failed run silently waits until sensor timeout.
2. The sensor cannot distinguish "failed once, will retry" from "failed permanently" without also polling the retry configuration and the retry attempt counter — a fragile contract that changes across Airflow versions.
3. Under datasets, none of this exists. The scheduler tracks the producer task's final state and emits the dataset event exactly once, on success. Retries that eventually succeed emit one event; runs that never succeed emit none.
4. The consumer DAG's first task starts at 05:10 (immediately after the retry succeeds), not at 05:30 (the naive cron offset that would have been used to "give ingest time to finish").
5. The mental cost is even bigger than the operational cost: the sensor requires the consumer team to reason about the producer's retry policy, an internal implementation detail. Datasets make the contract observable at the URI boundary and hide the retries entirely.

Output.

Time	Sensor version — what sensor sees	Dataset version — what scheduler sees
05:06	running	running
05:07	failed (race!)	failed
05:08	up_for_retry	up_for_retry
05:09	running	running
05:10	success — but sensor may already have failed	success → event emitted
Consumer start	fragile (0 or 05:10, depending on race)	05:10 (deterministic)

Rule of thumb. Datasets encode the "final-state, one-success-one-event" invariant directly. Sensors expose the full producer state machine, including intermediate states that leak into consumer logic. The senior signal is naming the retry-race explicitly during the migration story.

Worked example — DAG ownership boundaries across teams

Detailed explanation. A large-org anti-pattern: 12 DAGs owned by 4 teams, wired together with a spaghetti of ExternalTaskSensor calls that reach across team ownership. Team-ingest renames a task in its DAG; three consumer sensors break silently overnight. Datasets turn the URI into the contract; teams can rename internal tasks freely, and consumers only break when the URI itself changes — which is now a public API change that goes through review.

• The spaghetti. 12 DAGs × ~2 sensors per consumer ≈ 24 sensor edges, each hardcoded to a specific (dag_id, task_id) pair.
• The blast radius. Renaming a producer task breaks all sensors pointing at it. There's no compile-time or scheduler-time check; you find out at runtime.
• The dataset alternative. N producers × M consumers via K dataset URIs. Renames of internal tasks are invisible; URI changes are versioned like an API.

Question. Migrate a 4-team, 12-DAG spaghetti graph to datasets. Design the URI namespace, show the pattern for each team's producer, and quantify the coupling reduction.

Input.

Team	Producer DAG	Downstream consumers (via sensor)
ingest	ingest_orders_dag	3 consumers in analytics, finance, ml
ingest	ingest_customers_dag	2 consumers in analytics, marketing
analytics	transform_facts_dag	4 consumers in finance, reporting, ml, marketing
ml	feature_events_dag	1 consumer in ml_serving

Code.

# URI namespace convention — team-owned, storage-prefixed
INGEST_ORDERS      = Dataset("s3://ingest/orders")
INGEST_CUSTOMERS   = Dataset("s3://ingest/customers")
ANALYTICS_FACTS    = Dataset("snowflake://analytics/facts")
ML_FEATURE_EVENTS  = Dataset("s3://ml/feature_events")

# Team ingest — one outlet per producer DAG's final task
with DAG("ingest_orders_dag", schedule="0 5 * * *", ...) as d:
    ...
    final = PythonOperator(task_id="load", python_callable=lambda: None,
                           outlets=[INGEST_ORDERS])

with DAG("ingest_customers_dag", schedule="15 5 * * *", ...) as d:
    ...
    final = PythonOperator(task_id="load", python_callable=lambda: None,
                           outlets=[INGEST_CUSTOMERS])

# Team analytics — consumer of ingest, producer of facts
with DAG("transform_facts_dag",
         schedule=[INGEST_ORDERS, INGEST_CUSTOMERS],   # AND — wait for both
         ...) as d:
    ...
    final = PythonOperator(task_id="load", python_callable=lambda: None,
                           outlets=[ANALYTICS_FACTS])

# Team finance — consumer of facts
with DAG("finance_report_dag", schedule=[ANALYTICS_FACTS], ...) as d:
    ...

# Team ml — consumer of ingest_orders + feature_events
with DAG("ml_train_dag",
         schedule=[INGEST_ORDERS, ML_FEATURE_EVENTS],
         ...) as d:
    ...

Step-by-step explanation.

1. Design the URI namespace first, before touching any DAG. Convention: <storage>://<team>/<logical_name>. s3://ingest/orders says "team-ingest emits an orders dataset backed by S3." The convention is human, discoverable, and greppable across the codebase.
2. Each producer team adds outlets=[URI] to the final task of each producing DAG. That single line replaces every ExternalTaskSensor pointing at that DAG. Renaming internal tasks is now free — the outlet lives on whichever task is currently "final."
3. Each consumer team declares schedule=[URI_a, URI_b, ...] — the AND-across-datasets semantics. transform_facts_dag waits for both ingest/orders and ingest/customers before starting; when both events have arrived since the last consumer run, the scheduler fires.
4. Chains form naturally: transform_facts_dag is itself a producer (outlets=[ANALYTICS_FACTS]); finance_report_dag is a consumer of that dataset. Multi-hop lineage is implicit in the URI graph, not encoded in cross-DAG polling.
5. Coupling reduction: before, 24 hardcoded (dag_id, task_id) sensor edges; after, 4 URIs and every producer/consumer only knows the URIs it emits or consumes. Rename resilience is 100%; discoverability moves from tribal knowledge to the Datasets tab in the Airflow UI.

Output.

Metric	Before (sensors)	After (datasets)
Hardcoded (dag_id, task_id) edges	24	0
URI contracts	0	4
Renaming producer task	breaks N consumers	invisible
Renaming URI	not applicable	public API change (review)
Cross-team discoverability	Slack + tribal knowledge	Datasets tab
Multi-hop lineage	manual	implicit graph

Rule of thumb. Design the URI namespace before writing any producer/consumer code. The convention <storage>://<team>/<logical_name> is the load-bearing decision; get it right on day one and every future migration is trivial.

Senior interview question on the ExternalTaskSensor → dataset migration

A senior interviewer often opens with: "You inherit an Airflow deployment with 40 DAGs across 6 teams, all wired together with ExternalTaskSensor and hand-tuned cron offsets. Walk me through how you'd migrate to datasets, what URI namespace you'd design, and how you'd sequence the rollout without stopping the world."

Solution Using a phased URI-first migration plan

# Phase 1 — audit + URI namespace design
"""
Convention: <storage_scheme>://<owning_team>/<logical_name>
Examples:
    s3://ingest/orders
    s3://ingest/customers
    snowflake://analytics/dim_customer
    snowflake://analytics/fact_orders
    s3://ml/feature_events
    custom://finance/gl_close_status
"""

# Phase 2 — add outlets to producers WITHOUT removing sensors on consumers
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime

INGEST_ORDERS = Dataset("s3://ingest/orders")

with DAG("ingest_orders_dag",
         schedule="0 5 * * *",
         start_date=datetime(2026, 1, 1),
         catchup=False) as d:
    fetch = PythonOperator(task_id="fetch", python_callable=lambda: None)
    load  = PythonOperator(task_id="load",  python_callable=lambda: None,
                           outlets=[INGEST_ORDERS])
    fetch >> load

# Phase 3 — flip consumers one at a time
with DAG("transform_orders_dag",
         schedule=[INGEST_ORDERS],
         start_date=datetime(2026, 1, 1),
         catchup=False) as d:
    do = PythonOperator(task_id="do", python_callable=lambda: None)

# Phase 4 — after two weeks of parallel run, remove sensor DAGs

Phased rollout plan — 40 DAGs across 6 teams
============================================

Week 0 — Audit and namespace
  • Inventory every ExternalTaskSensor + TriggerDagRunOperator in the repo
  • Propose the URI namespace as an ADR (architecture decision record)
  • Review across all 6 teams; lock the convention

Week 1 — Producers add outlets
  • Every producing DAG adds outlets=[Dataset(URI)] on its final task
  • Sensors on consumers remain untouched — dual-write phase
  • Verify Datasets tab populates in the Airflow UI

Weeks 2–5 — Consumers migrate one team at a time
  • Team-A rewrites its consumers with schedule=[Dataset(URI)]
  • Sensor DAGs remain for team-A during shadow-mode week
  • After a week of parallel matching runs, delete the sensor DAGs

Week 6 — Cleanup
  • Remove all ExternalTaskSensor and TriggerDagRunOperator usage
  • Add freshness SLA alerts (dataset event age > threshold)
  • Document the URI catalog on the platform wiki

Step-by-step trace.

Phase	Producer side	Consumer side	Risk
0 — audit	grep the repo	grep the repo	none
1 — outlets added	outlets=[URI] on final task	unchanged (sensors still active)	zero — outlets are additive
2 — one consumer flips	unchanged	schedule=[URI] replaces cron + sensor	low — old sensor DAG still exists as fallback
3 — shadow week	both fire	both fire	can compare outputs
4 — delete sensor	outlets remain	sensor DAG deleted	low — datasets proven for a week
5 — repeat for each team	—	—	rolls forward one team at a time
6 — SLA alerts	—	—	freshness monitoring added

After phase 6, the deployment has 40 DAGs on datasets, zero ExternalTaskSensor usage, and a discoverable URI catalog in the Datasets tab. The migration takes 6–8 weeks and is fully reversible until phase 4 for each consumer.

Output:

Metric	Before	After
ExternalTaskSensor instances	~60	0
TriggerDagRunOperator instances	~15	0
Hand-tuned cron offsets	~40	0
URI contracts	0	~25
Worker slots held for polling	~60 during peaks	0
Cross-team discovery mechanism	Slack	Datasets tab

Why this works — concept by concept:

• URI as public contract — the namespace <storage>://<team>/<name> is a public API. Teams commit to emit it; teams subscribe to it. Renames go through review; internal task changes are invisible. The URI is the single load-bearing decision of the whole migration.
• Dual-write additive rollout — adding outlets to producers is additive; it never breaks anything. Consumers can migrate one at a time. Shadow-run for a week to confirm the dataset-triggered run and the sensor-triggered run produce identical outputs, then delete the sensor.
• Success-only event semantics — dataset events fire exactly on task success. Retry semantics, timeout races, and partial-failure states that plagued sensors all vanish. The senior signal is naming this invariant explicitly.
• Datasets tab as discoverability — the UI marketplace turns tribal knowledge into a discoverable catalog. Team-D can join the platform, see the URIs already available, and subscribe without any Slack thread.
• Cost — O(DAGs) engineering time during the phased rollout; O(1) operational cost afterward. Every future cross-DAG scheduling requirement is a one-line URI declaration on each side. The avoided cost of the next four "consumer broke because producer renamed a task" incidents pays for the migration in the first quarter.

ETL
Topic — etl
ETL problems on cross-DAG orchestration and scheduling

Practice →

Optimization Topic — optimization Optimization problems on Airflow throughput and scheduling

Practice →

---

2. Producer DAG — dataset outlets

outlets=[Dataset("s3://raw/orders")] on the emitting task turns any Airflow task into a data-aware event source

The mental model in one line: a producer task declares which datasets it emits via outlets=[Dataset(URI)]; on task-instance success, the Airflow scheduler emits a dataset event keyed by that URI; on any non-success terminal state, no event is emitted — the URI is a promise, outlets is where the promise is signed, and success is the only trigger. Every airflow dataset outlet interview question is a consequence of this one invariant.

The four rules of outlet declaration.

• Outlets live on tasks, not DAGs. The outlets=[...] argument goes on the specific operator that produces the dataset — usually the final write task. Putting outlets on an intermediate task (fetch, clean) fires the event too early, before the data is durable.
• One outlet, many downstreams. A single outlet URI can be consumed by any number of downstream DAGs. The producer never enumerates its consumers; the scheduler's routing does that for free.
• Success-only firing. The scheduler emits a dataset event exactly when the task instance's final state is success. Task failures, upstream_failed, skipped, and canceled states emit no event. Retries that eventually succeed emit one event.
• Multiple outlets on one task. outlets=[Dataset("s3://raw/orders"), Dataset("s3://raw/order_items")] is legal — one task producing two datasets in one shot fires both events on success. Downstream DAGs listening to either URI fire independently.

The URI naming grammar.

• Storage scheme prefix. s3://, gs://, azure://, snowflake://, bigquery://, hdfs://, or a custom scheme like logical:// or event://. The scheme is descriptive metadata; the scheduler treats it as opaque.
• Team or domain segment. s3://ingest/..., snowflake://analytics/..., custom://finance/.... The segment declares ownership.
• Logical name segment. orders, customers, fact_orders, feature_events. Keep it stable — this is what consumers pin to.
• Optional path components. s3://ingest/orders/dt=2026-06-22 is legal but usually a bad idea — putting the partition in the URI means every partition is a separate dataset, defeating the "one dataset, N runs" semantics. Keep partitions off the URI; encode partitions in task-level XCom or metadata.

The outlets ↔ inlets duality.

• outlets=[Dataset]. The task produces the dataset. Success emits the event.
• inlets=[Dataset]. The task consumes the dataset for lineage-tracking purposes only — it does not trigger the DAG. Inlets are documentation-in-code; the actual trigger is schedule=[Dataset] on the DAG.
• The scheduler cares about outlets. Inlets are for the Datasets tab lineage view. Don't confuse them; the senior interview probe is exactly this distinction.

What can go wrong at the producer side.

• Outlet on the wrong task. Placing outlets on the fetch task fires the event before the data is written. Consumers start reading empty S3 prefixes. Always put outlets on the final, durable-write task.
• Missing outlet on a rewritten task. Refactoring a DAG and forgetting to move the outlet to the new final task silently breaks every downstream consumer. The Datasets tab shows the URI hasn't been updated; the on-call gets a stale-data page.
• Overloaded outlet. One task producing 20 outlets is legal but usually a design smell — it means the task is doing too much and the failure blast radius is large. Split into 20 tasks each with its own outlet.
• Success on empty output. A task that succeeds without actually writing anything (early return on a data-not-ready condition) still emits the event. Handle this with a ShortCircuitOperator upstream, or guard the write inside the task and raise on empty.

Common interview probes on outlets.

• "Where does outlets live — on the DAG or on the task?" — required answer is task.
• "Do task failures emit dataset events?" — no; success-only.
• "Can one task emit multiple dataset events?" — yes; multiple outlets, one atomic firing.
• "What's the difference between outlets and inlets?" — outlets trigger downstream DAGs; inlets are lineage-only.
• "Where do you put the partition — in the URI or in metadata?" — in metadata; keep the URI stable.

Worked example — daily orders producer with outlets

Detailed explanation. The canonical producer DAG. A daily orders ingest DAG fetches raw orders from a source system, cleans them, and writes to s3://ingest/orders/. The DAG runs on cron (0 5 * * *). The single load task carries outlets=[Dataset("s3://ingest/orders")]. Show what the scheduler does on a normal run, on a retry-then-success, and on a permanent failure.

• Producer. ingest_orders_dag, cron 0 5 * * *, three tasks (fetch, clean, load).
• Outlet placement. load task carries the outlet — the durable-write step.
• Success emits. One event per successful load run.

Question. Write the producer DAG and enumerate the scheduler's behaviour under normal, retry-then-success, and permanent-failure scenarios.

Input.

Component	Value
DAG ID	ingest_orders_dag
Schedule	0 5 * * * (daily 05:00 UTC)
Outlet URI	s3://ingest/orders
Task placement	on load task
Retries	3 with 3-minute delay

Code.

from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

# URI — the cross-team contract; stable across renames of internal tasks
ORDERS_RAW = Dataset("s3://ingest/orders")

def _fetch_orders(**ctx):
    # Pull from source system into a local staging area
    print("fetched")

def _clean_orders(**ctx):
    # Deduplication, schema normalisation
    print("cleaned")

def _load_to_s3(**ctx):
    # Write parquet to s3://ingest/orders/dt=<logical_date>/
    # Raise on empty write to prevent success-with-no-data events
    n_rows = 12345
    if n_rows == 0:
        raise ValueError("no rows written — refusing to emit dataset event")
    print("loaded", n_rows, "rows")

default_args = {"retries": 3, "retry_delay": timedelta(minutes=3)}

with DAG(
    dag_id="ingest_orders_dag",
    start_date=datetime(2026, 1, 1),
    schedule="0 5 * * *",
    catchup=False,
    default_args=default_args,
    tags=["ingest", "orders"],
) as dag:

    fetch = PythonOperator(task_id="fetch_orders", python_callable=_fetch_orders)
    clean = PythonOperator(task_id="clean_orders", python_callable=_clean_orders)
    load  = PythonOperator(
        task_id="load_to_s3",
        python_callable=_load_to_s3,
        outlets=[ORDERS_RAW],           # <-- outlet on the durable-write task
    )

    fetch >> clean >> load

Step-by-step explanation.

1. The three-task shape (fetch, clean, load) reflects the standard ingest pipeline. The outlet is placed only on load because that's the step that produces the durable artifact consumers pin against; putting it on fetch or clean would fire the event before the S3 write completes.
2. Normal run at 05:00 UTC: the scheduler triggers the DAG on cron; fetch → clean → load execute serially; load reaches success state; scheduler emits one dataset event for s3://ingest/orders with timestamp 05:07 (or whenever load completes). Any DAG with schedule=[ORDERS_RAW] gets enqueued.
3. Retry-then-success run: load fails on attempt 1 (say, an S3 transient), enters up_for_retry, retries at 05:10, succeeds. The scheduler emits exactly one dataset event at 05:10. The event timestamp is the final success time, not the first-attempt time.
4. Permanent failure run: load fails all three attempts and enters failed state. No dataset event is emitted. Every downstream consumer stays quiescent. Freshness SLAs (covered in section 5) are the mechanism to alert on this case; the scheduler itself is intentionally silent.
5. The guard if n_rows == 0: raise ... is the "success on empty" defence — a task that would otherwise return normally on zero rows would emit an event promising "data available" when there is none. Raising forces the task into failed state and suppresses the event. Consumers stay stopped.

Output.

Scenario	Final load state	Event emitted?	Consumer runs?
Normal	success at 05:07	yes (t=05:07)	yes, starts 05:07
Retry-then-success	success at 05:10	yes (t=05:10)	yes, starts 05:10
Permanent failure	failed	no	no (silent)
Success on empty (guarded)	failed	no	no (guarded)
Success on empty (unguarded)	success	yes	yes — bug

Rule of thumb. Always guard your producer's final task against "success on empty" — raise, don't return, when the write set is empty. The one line of guard code prevents an entire class of "consumer ran on stale/empty data" incidents.

Worked example — multiple outlets from one task

Detailed explanation. A single ingest task that writes orders and order_items in one atomic transaction against a source system. Both outputs land in S3 as parquet at the same instant. Declaring both outlets on the one task fires both events simultaneously; downstream consumers listening to either URI (or both) fire independently.

• Producer. ingest_orders_and_items_dag, one load task producing two S3 prefixes.
• Outlets. [Dataset("s3://ingest/orders"), Dataset("s3://ingest/order_items")].
• Consumers. transform_orders_dag listens to orders only; transform_order_items_dag listens to order_items only; report_orders_full_dag listens to both (AND).

Question. Write the producer DAG with two outlets on one task and describe what each consumer sees.

Input.

Component	Value
Producer task	load_orders_and_items
Outlet 1	s3://ingest/orders
Outlet 2	s3://ingest/order_items
Consumer A	schedule=[Dataset("s3://ingest/orders")]
Consumer B	schedule=[Dataset("s3://ingest/order_items")]
Consumer C	schedule=[Dataset("s3://ingest/orders"), Dataset("s3://ingest/order_items")]

Code.

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

ORDERS       = Dataset("s3://ingest/orders")
ORDER_ITEMS  = Dataset("s3://ingest/order_items")

def _load_pair(**ctx):
    # Single transaction against source; single atomic write to S3.
    # Raises on any partial-write condition.
    print("loaded orders and order_items atomically")

with DAG(
    dag_id="ingest_orders_and_items_dag",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    load = PythonOperator(
        task_id="load_orders_and_items",
        python_callable=_load_pair,
        outlets=[ORDERS, ORDER_ITEMS],   # <-- two outlets on one task
    )

Step-by-step explanation.

1. On task success, the scheduler emits both dataset events atomically — one for each URI in outlets. There is no ordering guarantee between the two events at the microsecond level, but from a scheduling standpoint they are treated as a single logical transaction.
2. Consumer A (schedule=[ORDERS]) receives the s3://ingest/orders event and enqueues its DAG run. It is not delayed by the order_items event — the scheduler processes each dataset independently.
3. Consumer B (schedule=[ORDER_ITEMS]) receives the s3://ingest/order_items event and enqueues its DAG run. It runs in parallel with consumer A.
4. Consumer C (schedule=[ORDERS, ORDER_ITEMS]) uses AND semantics — it fires only when both events have been received since its last run. Because both events fire in the same producer task success, consumer C also fires immediately after this run.
5. Emitting multiple outlets from one task is the right choice when the writes are atomic (single transaction). When the writes are logically separate — say, orders written at 05:07 and order_items written at 06:12 — split them into two tasks, each with its own outlet, so consumers of order_items don't wait for orders and vice versa.

Output.

Consumer	Schedule	Fires at
A (orders only)	[ORDERS]	05:07 (immediately)
B (order_items only)	[ORDER_ITEMS]	05:07 (immediately)
C (both AND)	[ORDERS, ORDER_ITEMS]	05:07 (both events already received)

Rule of thumb. Emit multiple outlets from one task only when the writes are genuinely atomic. If the writes are logically separate — different completion times, different failure semantics — split them into two tasks so consumers of one don't wait for the other.

Worked example — outlets on TaskFlow decorator API

Detailed explanation. The TaskFlow API is the modern Airflow way to define tasks — Python functions decorated with @task. Outlets work identically on TaskFlow tasks; the syntax is a outlets=[...] kwarg on the @task(...) decorator. Interviewers increasingly probe this because TaskFlow is the recommended pattern in Airflow 2.3+.

• TaskFlow decorator. @task(outlets=[Dataset(URI)]).
• Equivalent to PythonOperator. PythonOperator(..., outlets=[Dataset(URI)]).
• Same success-only semantics. Nothing about outlet behaviour differs between the two APIs.

Question. Rewrite the daily-orders producer using the TaskFlow API and show the outlet placement.

Input.

Component	Value
API	TaskFlow (@dag, @task)
Producer function	load_to_s3
Outlet URI	s3://ingest/orders

Code.

from airflow.decorators import dag, task
from airflow import Dataset
from datetime import datetime

ORDERS_RAW = Dataset("s3://ingest/orders")

@dag(
    dag_id="ingest_orders_taskflow_dag",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["ingest", "orders", "taskflow"],
)
def ingest_orders_taskflow():

    @task
    def fetch():
        return {"rows_fetched": 12345}

    @task
    def clean(payload: dict) -> dict:
        return {"rows_cleaned": payload["rows_fetched"]}

    @task(outlets=[ORDERS_RAW])            # <-- outlet on the final task
    def load(payload: dict) -> None:
        if payload["rows_cleaned"] == 0:
            raise ValueError("no rows to load")
        print("loaded", payload["rows_cleaned"])

    load(clean(fetch()))

ingest_orders_taskflow()

Step-by-step explanation.

1. The @dag decorator replaces the with DAG(...) as d: context manager. The DAG-level arguments (schedule, start_date, catchup) are decorator kwargs. The function body defines the task graph.
2. Each @task decorator turns a Python function into an Airflow task. The outlets=[...] kwarg on the decorator is where the dataset outlet declaration lives. The rest of the outlet semantics (success-only, atomic firing) is identical to the PythonOperator version.
3. The function-call syntax load(clean(fetch())) inside the DAG body wires up the task dependencies — fetch → clean → load — through Airflow's XCom-based data passing. The final call load(...) is where the outlet-carrying task is materialised.
4. Success-only semantics are enforced by the scheduler exactly as in the PythonOperator case. Raising in the function body puts the task into failed state; returning normally puts it into success and emits the event.
5. Both APIs coexist; teams pick based on style. The senior interview signal is knowing the outlet syntax works identically in either — it's not a TaskFlow limitation and not a TaskFlow feature, it's a scheduler-layer primitive.

Output.

API	Outlet syntax	Behaviour
Classic PythonOperator	PythonOperator(outlets=[Dataset(URI)])	success emits event
TaskFlow @task	@task(outlets=[Dataset(URI)])	success emits event

Rule of thumb. Outlets are a scheduler-level primitive; the API you use to define tasks is a stylistic choice. TaskFlow is the recommended pattern for new DAGs; outlets work identically in both. Never claim outlets don't work with TaskFlow — that answer betrays a stale mental model.

Senior interview question on outlet placement + naming

A senior interviewer might ask: "You're designing the URI namespace and outlet placement for a producer that ingests 5 tables (orders, order_items, customers, products, addresses) from a source database in one Airflow DAG. Walk me through where you'd place the outlets, how you'd design the URI names, and what happens if the source database only partially loads on a given day."

Solution Using per-table load tasks each with a single outlet

from airflow.decorators import dag, task
from airflow import Dataset
from datetime import datetime

# URI namespace — one per table, team-owned
ORDERS      = Dataset("s3://ingest/orders")
ORDER_ITEMS = Dataset("s3://ingest/order_items")
CUSTOMERS   = Dataset("s3://ingest/customers")
PRODUCTS    = Dataset("s3://ingest/products")
ADDRESSES   = Dataset("s3://ingest/addresses")

@dag(
    dag_id="ingest_orders_domain_dag",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["ingest", "orders_domain"],
)
def ingest_orders_domain():

    @task
    def snapshot_source():
        # Snapshot the source DB to a staging area; one atomic snapshot
        return {"snapshot_id": "20260622-050000"}

    @task(outlets=[ORDERS])
    def load_orders(snapshot: dict):
        _load_table(snapshot["snapshot_id"], "orders", require_nonempty=True)

    @task(outlets=[ORDER_ITEMS])
    def load_order_items(snapshot: dict):
        _load_table(snapshot["snapshot_id"], "order_items", require_nonempty=True)

    @task(outlets=[CUSTOMERS])
    def load_customers(snapshot: dict):
        _load_table(snapshot["snapshot_id"], "customers", require_nonempty=False)

    @task(outlets=[PRODUCTS])
    def load_products(snapshot: dict):
        _load_table(snapshot["snapshot_id"], "products", require_nonempty=False)

    @task(outlets=[ADDRESSES])
    def load_addresses(snapshot: dict):
        _load_table(snapshot["snapshot_id"], "addresses", require_nonempty=False)

    snap = snapshot_source()
    load_orders(snap)
    load_order_items(snap)
    load_customers(snap)
    load_products(snap)
    load_addresses(snap)


def _load_table(snapshot_id: str, table: str, require_nonempty: bool):
    n = _write_parquet(snapshot_id, table)
    if require_nonempty and n == 0:
        raise ValueError(f"{table} loaded 0 rows on snapshot {snapshot_id}")

def _write_parquet(snapshot_id: str, table: str) -> int:
    # Actual write to s3://ingest/<table>/dt=<logical_date>/
    return 12345

ingest_orders_domain()

Step-by-step trace.

Task	Outlet	Nonempty guard	Behaviour on partial load
snapshot_source	none	no	just snapshots
load_orders	ORDERS	yes	fails on empty; no event; orders consumers stay quiescent
load_order_items	ORDER_ITEMS	yes	fails on empty; no event
load_customers	CUSTOMERS	no	success on empty; event fires; downstream may run on stale
load_products	PRODUCTS	no	success on empty; event fires
load_addresses	ADDRESSES	no	success on empty; event fires

The design gives each table its own outlet; consumers of orders don't wait for customers to finish loading. Partial-load days (e.g. source DB skipped the customers table) fire some events but not others; consumers of the missing tables stay quiescent, and freshness SLAs catch the stale state.

Output:

Table	URI	Consumers today	Downstream fires?
orders	s3://ingest/orders	3	yes (event fired)
order_items	s3://ingest/order_items	2	yes
customers	s3://ingest/customers	4	yes (or stays quiet if load fails)
products	s3://ingest/products	2	yes
addresses	s3://ingest/addresses	1	yes

Why this works — concept by concept:

• One outlet per logical dataset — placing each outlet on its own task means partial-load days emit some events and not others. Consumers of the healthy tables run; consumers of the failed table stay quiet. Fine-grained blast radius.
• Nonempty guard — for critical tables (orders, order_items), the guard prevents "success on zero rows" from firing the event and starting downstream on stale data. For optional tables (products), the guard is omitted because an occasional empty load is acceptable.
• URI namespace — s3://ingest/<table> is stable across renames and refactors. If the source table changes name upstream, the URI stays; the load task's internal implementation changes but the contract does not.
• Snapshot upstream, load parallel — snapshot_source runs first and provides a snapshot_id XCom that all load tasks consume. This keeps source-system reads atomic while allowing S3 writes to parallelise.
• Cost — one extra task per table (versus one giant load task), which is O(N) in table count but with tiny per-task overhead. The blast-radius win far outweighs the coordination cost.

ETL
Topic — etl
ETL problems on Airflow producer patterns

Practice →

SQL Topic — sql SQL problems on incremental ingest and event ordering

Practice →

---

3. Consumer DAG — schedule on dataset

dag = DAG(schedule=[Dataset(...)]) is the event-driven trigger — no cron, no sensor, AND-across-multiple

The mental model in one line: passing a list of Dataset objects to schedule= replaces every cron + sensor combination — the DAG runs when all datasets in the list have received at least one event since the last run, the AND semantics is non-configurable in the base primitive, and DatasetOrTimeSchedule is the only escape hatch for combining datasets with a cron backstop. Every airflow schedule dataset interview question is a consequence of this shape.

The four rules of consumer scheduling.

• Datasets in a list = AND. schedule=[Dataset(a), Dataset(b)] fires only when both have received an event since the last DAG run. There is no OR in the base primitive.
• "Since last run" is the counter. The scheduler tracks per-DAG event consumption. A dataset that fires twice between two consumer runs counts as one — the consumer catches up on the second event's data when it next runs.
• One-time triggering. Once all datasets in the list have events, the consumer fires exactly once. It doesn't fire per-event; it fires per-quorum. This is important for AND workloads that get multiple upstream events between runs.
• No cron in the base form. schedule=[Dataset(a)] is purely event-driven. To combine with cron (say, "run when dataset arrives, but at least daily"), use DatasetOrTimeSchedule(timetable=..., datasets=[...]).

How the AND semantics works in practice.

• State tracked per DAG. The scheduler stores "last event timestamp consumed" per (consumer_dag, dataset). Each dataset event advances that timestamp.
• Fire condition. All datasets in the schedule list have advanced past the last consumer-run timestamp. When true, enqueue a run.
• Reset on fire. When the consumer runs, its "last consumed" markers all update. The next fire condition is another quorum of new events.
• Missed events collapse. If dataset A fires 3 times before the consumer next runs, those collapse into one — the consumer sees the latest state, not each intermediate one. This is by design: consumers are batch-latest, not per-event.

The DatasetOrTimeSchedule compound.

• Purpose. Fire on any dataset event or on a cron schedule — whichever comes first. Useful for "run when data arrives, but always run daily even if data is stale."
• Shape. DatasetOrTimeSchedule(timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"), datasets=[Dataset(URI)]).
• Semantics. The consumer fires either when the dataset event lands or when the cron time arrives. Both can fire on the same day — one dataset-triggered run at 05:07, one cron-triggered run at 06:00.
• The gotcha. If the dataset event fires close to the cron time, you can get two runs within minutes. Guard against duplicate work with idempotency on the consumer side.

The single-dataset case — the simplest form.

• Shape. schedule=[Dataset("s3://ingest/orders")].
• Fire condition. One event since last run.
• Behaviour. Consumer runs exactly once per producer success. Most common form; the entry-level Airflow datasets pattern.

The multi-dataset AND case.

• Shape. schedule=[Dataset(A), Dataset(B), Dataset(C)].
• Fire condition. All three have events since last run.
• Behaviour. Consumer waits for the slowest producer, then fires. If A fires daily at 05:00, B at 06:00, C at 07:00 — the consumer fires at 07:00 (the moment the last quorum event arrives).

Common interview probes on consumer scheduling.

• "What semantics does a list of datasets have — AND or OR?" — required answer: AND.
• "How do you combine dataset triggers with a cron backstop?" — DatasetOrTimeSchedule.
• "If the producer fires 3 times before the consumer runs, how many consumer runs happen?" — one; events collapse.
• "Can a consumer have both schedule_interval and dataset triggers?" — only via DatasetOrTimeSchedule in modern Airflow; the old schedule_interval alone is either/or.
• "What's the DAG run's logical date when triggered by a dataset event?" — the dataset event's timestamp (2.7+); earlier versions used consumer start time.

Worked example — single-dataset consumer

Detailed explanation. The simplest consumer. transform_orders_dag runs when s3://ingest/orders receives an event. Show the DAG, the fire behaviour on a normal producer success, and the state-tracking mental model.

• Consumer. transform_orders_dag, schedule=[Dataset("s3://ingest/orders")].
• Producer. ingest_orders_dag emits the dataset once daily.
• Fire timing. Consumer starts within seconds of the producer's load success.

Question. Write the consumer DAG and describe what the scheduler does when the producer emits the dataset event.

Input.

Component	Value
Consumer DAG ID	transform_orders_dag
Schedule	[Dataset("s3://ingest/orders")]
Consumer tasks	stage, transform, load_dw

Code.

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

ORDERS_RAW = Dataset("s3://ingest/orders")

with DAG(
    dag_id="transform_orders_dag",
    schedule=[ORDERS_RAW],               # <-- data-aware trigger, no cron
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["analytics", "orders"],
) as dag:

    stage = PythonOperator(task_id="stage",   python_callable=lambda: None)
    tx    = PythonOperator(task_id="transform", python_callable=lambda: None)
    load  = PythonOperator(task_id="load_dw", python_callable=lambda: None)

    stage >> tx >> load

Step-by-step explanation.

1. The consumer DAG declares its trigger as schedule=[ORDERS_RAW]. No cron string, no ExternalTaskSensor, no execution_delta arithmetic. The scheduler now watches for events keyed to s3://ingest/orders.
2. Internal state tracked by the scheduler: for (transform_orders_dag, s3://ingest/orders), store the "last consumed event timestamp." On DAG creation, this is set to the DAG's start_date so historical events don't fire an immediate catch-up run (unless catchup=True).
3. When the producer's load task reaches success at time T, the scheduler records a new dataset event at T. It checks every DAG whose schedule list contains this URI; for each such DAG, it evaluates the fire condition (all datasets in the list have events after the last consumer-run's timestamp).
4. transform_orders_dag's schedule list has one dataset; that dataset has one new event; the fire condition is true. The scheduler enqueues a new DAG run for transform_orders_dag with logical date T (the event timestamp).
5. The DAG run starts almost immediately — no polling, no sensor wait. The consumer's stage task begins within scheduler-loop latency (typically seconds) of the producer's success.

Output.

Event	Time	Consumer state
Producer load success	05:07	dataset event emitted
Scheduler evaluates	05:07	fire condition true → enqueue run
Consumer DAG run starts	05:07 + few seconds	stage running
Consumer completes	05:15	run marked success; last consumed updates to 05:07
No more events until tomorrow		consumer stays quiescent

Rule of thumb. For the single-producer / single-consumer case, schedule=[Dataset(URI)] is the entire declaration. There's no cron, no sensor, no glue. If you find yourself writing more than one line of scheduling code on the consumer side, you're either doing multi-dataset AND or combining with cron via DatasetOrTimeSchedule.

Worked example — multi-dataset AND consumer

Detailed explanation. A revenue reporting DAG that needs orders, customers, and payments to all have fresh data before it runs. Declared as schedule=[Dataset(A), Dataset(B), Dataset(C)]. Producers emit at different times; the consumer fires only when the last quorum event arrives.

• Consumer. daily_revenue_report_dag, three datasets in schedule list.
• Producer emission times. orders at 05:07, customers at 05:12, payments at 05:45.
• Consumer fires. At 05:45, when the last of the three events arrives.

Question. Write the consumer and trace the scheduler's state at each producer emission.

Input.

Producer	Dataset URI	Emit time
ingest_orders_dag	s3://ingest/orders	05:07
ingest_customers_dag	s3://ingest/customers	05:12
ingest_payments_dag	s3://ingest/payments	05:45

Code.

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

ORDERS    = Dataset("s3://ingest/orders")
CUSTOMERS = Dataset("s3://ingest/customers")
PAYMENTS  = Dataset("s3://ingest/payments")

with DAG(
    dag_id="daily_revenue_report_dag",
    schedule=[ORDERS, CUSTOMERS, PAYMENTS],   # AND of all three
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["reporting", "revenue"],
) as dag:

    build_report = PythonOperator(
        task_id="build_report",
        python_callable=lambda: None,
    )

Step-by-step explanation.

1. The consumer DAG's schedule list contains three datasets. The scheduler tracks a (consumer_dag, dataset) pair for each — three internal counters keyed to this consumer.
2. At 05:07 the orders event fires. The scheduler updates the (daily_revenue_report_dag, orders) counter. It evaluates the fire condition: orders has a new event, customers does not, payments does not. The AND is not satisfied; no run is enqueued.
3. At 05:12 the customers event fires. Counter updates. Fire condition: orders yes, customers yes, payments no. Still not satisfied.
4. At 05:45 the payments event fires. Counter updates. Fire condition: all three yes. The scheduler enqueues a run for daily_revenue_report_dag with logical date 05:45 (the timestamp of the last event that completed the quorum).
5. When the consumer's DAG run starts and completes, all three "last consumed" markers reset. The next quorum requires another event from each of the three producers.

Output.

Time	Event	Fire condition	Consumer state
05:07	orders fires	1/3	quiescent
05:12	customers fires	2/3	quiescent
05:45	payments fires	3/3	run enqueued
05:47	run starts	—	running
05:58	run completes	—	all counters reset

Rule of thumb. AND semantics is what you almost always want for multi-source consumers. Design so the slowest producer determines the consumer's fire time; the consumer never runs on partial data. Latency is bounded by the slowest producer; correctness is guaranteed by the AND.

Worked example — DatasetOrTimeSchedule cron backstop

Detailed explanation. A daily reporting DAG that must fire when a dataset event arrives, but also must fire at 07:00 UTC regardless — even if the dataset is stale. This is the pattern for regulatory-report DAGs that have a hard deadline. DatasetOrTimeSchedule combines the two.

• Consumer. regulatory_report_dag.
• Behaviour. Fire on s3://finance/gl event OR at 07:00 UTC, whichever comes first.
• Purpose. Guarantee the report ships every day; prefer fresh data when available.

Question. Write the consumer with DatasetOrTimeSchedule and show the fire behaviour on three scenarios: dataset arrives before 07:00, dataset arrives after 07:00, dataset never arrives.

Input.

Component	Value
Dataset	s3://finance/gl
Cron backstop	0 7 * * * (07:00 UTC)
Timezone	UTC

Code.

from airflow import DAG, Dataset
from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.operators.python import PythonOperator
from datetime import datetime

GL_DATASET = Dataset("s3://finance/gl")

schedule = DatasetOrTimeSchedule(
    timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"),
    datasets=[GL_DATASET],
)

with DAG(
    dag_id="regulatory_report_dag",
    schedule=schedule,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["finance", "regulatory"],
) as dag:

    build_report = PythonOperator(
        task_id="build_report",
        python_callable=lambda: None,
    )

Step-by-step explanation.

1. DatasetOrTimeSchedule is a timetable — a compound scheduling object that combines a cron-style timetable with a dataset trigger. The scheduler evaluates both; either can fire the DAG.
2. Scenario A: dataset event arrives at 05:30. The scheduler enqueues a run at 05:30. When 07:00 UTC arrives, the scheduler checks whether a cron run is due — since the dataset run already covered the current logical window, the cron does not re-fire (deduplication).
3. Scenario B: dataset event does not arrive by 07:00. The scheduler fires the cron run at 07:00. The consumer runs on whatever state the source data is currently in (possibly stale — the freshness alert catches this).
4. Scenario C: dataset event arrives at 08:00 (after the cron fired). The scheduler enqueues another run at 08:00, this time triggered by the dataset. So both fired — the 07:00 cron run and the 08:00 dataset run.
5. The gotcha in scenario C is that the consumer runs twice. Design the consumer to be idempotent — running twice must not produce duplicate side effects. The standard pattern is a merge/upsert target rather than an append, plus a check for "was there already a fresher run today?" and a short-circuit.

Output.

Scenario	Dataset event	Cron fires at	Actual runs
A: fresh data	05:30	07:00 (skipped)	1 run at 05:30
B: stale data	never	07:00	1 run at 07:00 (stale)
C: late data	08:00	07:00	2 runs — 07:00 (stale) + 08:00 (fresh)

Rule of thumb. DatasetOrTimeSchedule is the correct primitive for "prefer event, guarantee cron backstop." Always design the consumer to be idempotent — the scenario-C double-fire is real and depends on producer timing you don't control.

Senior interview question on consumer schedule shape selection

A senior interviewer might ask: "You have four DAGs to design consumer schedules for: (1) a real-time transform that must fire on any producer event, (2) a daily report that needs three specific datasets to be fresh, (3) a compliance job that must fire once per day whether data is fresh or not, and (4) an on-demand refresh triggered by a manual dataset event. Walk me through each schedule declaration and why."

Solution Using four schedule shapes matched to four requirements

from airflow import DAG, Dataset
from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable
from airflow.operators.python import PythonOperator
from datetime import datetime

# 1. Real-time transform — fires on any producer event
ORDERS = Dataset("s3://ingest/orders")
with DAG(dag_id="realtime_transform_dag",
         schedule=[ORDERS],
         start_date=datetime(2026, 1, 1),
         catchup=False) as d1:
    PythonOperator(task_id="do", python_callable=lambda: None)

# 2. Daily report — three datasets, AND semantics
CUSTOMERS = Dataset("s3://ingest/customers")
PAYMENTS  = Dataset("s3://ingest/payments")
with DAG(dag_id="daily_revenue_report_dag",
         schedule=[ORDERS, CUSTOMERS, PAYMENTS],
         start_date=datetime(2026, 1, 1),
         catchup=False) as d2:
    PythonOperator(task_id="build", python_callable=lambda: None)

# 3. Compliance job — cron backstop, dataset preferred
GL_DATASET = Dataset("s3://finance/gl")
d3_sched = DatasetOrTimeSchedule(
    timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"),
    datasets=[GL_DATASET],
)
with DAG(dag_id="regulatory_report_dag",
         schedule=d3_sched,
         start_date=datetime(2026, 1, 1),
         catchup=False) as d3:
    PythonOperator(task_id="report", python_callable=lambda: None)

# 4. On-demand — manual-trigger-only via a synthetic dataset
MANUAL_TRIGGER = Dataset("event://ops/manual_refresh")
with DAG(dag_id="manual_refresh_dag",
         schedule=[MANUAL_TRIGGER],
         start_date=datetime(2026, 1, 1),
         catchup=False) as d4:
    PythonOperator(task_id="refresh", python_callable=lambda: None)

# Producer for MANUAL_TRIGGER — a tiny "signal" DAG that ops can trigger manually
with DAG(dag_id="ops_signal_manual_refresh",
         schedule=None,                # manual only
         start_date=datetime(2026, 1, 1),
         catchup=False) as d5:
    signal = PythonOperator(
        task_id="emit",
        python_callable=lambda: None,
        outlets=[MANUAL_TRIGGER],
    )

Step-by-step trace.

DAG	Schedule shape	Fire condition	Reasoning
realtime_transform_dag	[Dataset(A)]	one event	simplest; real-time transform on new orders
daily_revenue_report_dag	[Dataset(A), Dataset(B), Dataset(C)]	AND all three	multi-source; wait for slowest producer
regulatory_report_dag	DatasetOrTimeSchedule(cron, [Dataset])	event OR cron	hard deadline; prefer fresh, guarantee daily
manual_refresh_dag	[Dataset("event://ops/manual_refresh")]	one event from signal DAG	manual trigger via synthetic dataset event

The four shapes cover 95% of real-world consumer needs. The custom event:// scheme in shape 4 is the accepted convention for "synthetic" datasets that don't correspond to physical storage — the URI is just a stable event key.

Output:

Shape	Example use case	Schedule expression
Single-dataset	real-time transform	schedule=[Dataset(A)]
Multi-dataset AND	multi-source report	schedule=[Dataset(A), Dataset(B), Dataset(C)]
Dataset OR cron	compliance job with hard deadline	schedule=DatasetOrTimeSchedule(...)
Synthetic dataset	manual-trigger workflow	schedule=[Dataset("event://ops/manual")]

Why this works — concept by concept:

• Match shape to requirement — the four shapes are canonical; every consumer requirement maps to one. Interview-grade answer is naming the shape and defending the choice with the fire condition.
• AND is non-negotiable — passing multiple datasets in a list is AND, not OR. If OR semantics is required, use two separate consumer DAGs (each listening to one dataset) that share task code via a Python module.
• Synthetic datasets for manual triggers — the event:// scheme (or custom://, logical://) is the convention for URIs that don't correspond to physical storage. The scheduler doesn't care; the URI is just an event key.
• Idempotency guard on cron backstop — the DatasetOrTimeSchedule double-fire scenario is real. Design consumers to be idempotent (upsert not append; check-then-write). This is the senior detail interviewers probe for.
• Cost — zero runtime cost; scheduling shape is a declaration only. The engineering cost is choosing the correct shape once per consumer DAG. Choose right, ship, forget.

ETL
Topic — etl
ETL problems on multi-source consumer scheduling

Practice →

Optimization Topic — optimization Optimization problems on scheduling latency and freshness

Practice →

---

4. Cross-team lineage + dataset UI

The Datasets tab is the marketplace — every producer, every consumer, every event, discoverable from one page

The mental model in one line: the Airflow UI's Datasets tab renders every declared dataset URI as a discoverable catalog entry, showing the producer DAG(s) that emit it, the consumer DAG(s) that consume it, and the recent event history — the URI becomes the cross-team contract, the tab becomes the marketplace, and the naming convention you pick on day one determines whether the catalog is a searchable index or an unreadable pile. Every senior cross-dag dependency interview question in 2026 assumes datasets, and every datasets question assumes the URI namespace is designed thoughtfully.

What the Datasets tab actually shows.

• URI catalog. Every URI declared by any producer's outlets=[...] or any consumer's schedule=[...]. The tab is the source-of-truth registry.
• Producer / consumer counts. Per URI: how many DAGs emit it, how many DAGs consume it. A URI with 3 producers is usually a design smell; a URI with many consumers is a healthy asset.
• Event history. Recent dataset events with timestamps. Useful for "when did this last fire?" incident investigation.
• Manual event injection. The UI supports injecting a synthetic dataset event for debugging or manual triggering — the "trigger downstream now" affordance.
• Downstream graph. Click through to see the DAG-level lineage graph rooted at a URI.

The URI naming convention that scales.

• <storage>://<team>/<logical_name> — the canonical pattern.
• s3://ingest/orders, snowflake://analytics/fact_orders, custom://finance/gl_close_status — real examples in production shops.
• Stability. The URI is a public contract. Once shipped, changing it is a breaking API change; treat renames like a database schema migration (deprecation window, dual-emit, cutover).
• No environment prefix. Don't put dev, stg, prd in the URI. Environments are handled at deployment scope, not in the URI string.
• No partition suffix. Don't put dt=2026-06-22 in the URI. Partitions belong in metadata (XCom, task params, logical_date), not in the URI.

Team ownership on datasets.

• The producing team owns the URI. The team whose DAG has outlets=[URI] is the owner. Owner responsibilities: honour the emission cadence, communicate breaking changes, respond to freshness alerts.
• The consuming team subscribes. Consumer teams pin to the URI and communicate breaking-change expectations to the owner. Consumer teams do not own the URI.
• Multi-team URIs are dangerous. A URI produced by two teams (e.g. s3://raw/orders emitted by both ingest_v1 and ingest_v2 during migration) creates ambiguity about who's on the hook for outages. Design URIs to have one producer team.
• The catalog view enables self-service subscription. Team-D joins the platform, browses the Datasets tab, finds s3://ingest/orders, and writes a consumer DAG — no Slack thread, no cross-team ticket.

The lineage graph.

• Two-hop is the common shape. Producer → dataset → consumer. Most real graphs go three-hop or more (ingest → analytics → reporting), and the datasets tab visualises these.
• The graph is derived, not declared. You declare outlets on producers and datasets in consumer schedules; the graph is the transitive closure of "who emits, who consumes."
• The graph is dynamic. Add a new consumer, the graph updates automatically. Remove a producer, the URI still exists (with a "no active producer" warning) until you also remove references from consumers.

Cross-team communication patterns.

• URI ADR. Each team publishes an ADR (architecture decision record) declaring the URIs it produces, the emission cadence, and the schema of the underlying data. The ADR is the durable contract; the URI is the runtime pointer.
• Freshness SLA. The producing team commits to "URI X will fire at least once every 24 hours." Consumers rely on this SLA; alerts fire when it's violated.
• Schema evolution. The URI is stable; the schema underneath can evolve. Additive changes (new columns) are transparent; breaking changes (renamed columns, dropped fields) require a versioned URI: s3://ingest/orders_v2.

Common interview probes on cross-team lineage.

• "How do you handle a URI that two teams want to emit?" — you don't; one team owns; the other consumes and re-emits with a different URI.
• "How do you rename a URI without breaking consumers?" — dual-emit for a deprecation window, then cut over.
• "How do you communicate schema changes downstream?" — a schema registry paired with the URI, or an additive-only policy backed by CI.
• "What's the difference between outlets and the Datasets tab?" — outlets declare datasets; the tab catalogs them by aggregating declarations across all DAGs.
• "Can you see who consumes a dataset from the tab?" — yes, that's the discoverability affordance.

Worked example — 3-team lineage graph

Detailed explanation. Three teams — ingest, analytics, reporting — each with their own producer/consumer DAGs. Ingest produces raw datasets; analytics consumes raw and produces dimensional datasets; reporting consumes dimensional datasets. The lineage graph is a three-hop chain visible in the Datasets tab.

• Team ingest. Produces s3://ingest/orders and s3://ingest/customers.
• Team analytics. Consumes those, produces snowflake://analytics/fact_orders and snowflake://analytics/dim_customer.
• Team reporting. Consumes the analytics datasets, produces snowflake://reporting/daily_revenue.

Question. Write all six DAGs with the correct outlet/schedule declarations. Describe what the Datasets tab shows.

Input.

Team	DAG	Datasets consumed	Datasets produced
ingest	ingest_orders_dag	none (cron)	s3://ingest/orders
ingest	ingest_customers_dag	none (cron)	s3://ingest/customers
analytics	build_fact_orders_dag	s3://ingest/orders	snowflake://analytics/fact_orders
analytics	build_dim_customer_dag	s3://ingest/customers	snowflake://analytics/dim_customer
reporting	daily_revenue_dag	both analytics datasets	snowflake://reporting/daily_revenue

Code.

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

# URIs — cross-team contracts
ORDERS_RAW  = Dataset("s3://ingest/orders")
CUSTOMERS_RAW = Dataset("s3://ingest/customers")
FACT_ORDERS   = Dataset("snowflake://analytics/fact_orders")
DIM_CUSTOMER  = Dataset("snowflake://analytics/dim_customer")
DAILY_REVENUE = Dataset("snowflake://reporting/daily_revenue")

# Team ingest
with DAG("ingest_orders_dag", schedule="0 5 * * *",
         start_date=datetime(2026, 1, 1), catchup=False) as d:
    PythonOperator(task_id="load", python_callable=lambda: None,
                   outlets=[ORDERS_RAW])

with DAG("ingest_customers_dag", schedule="15 5 * * *",
         start_date=datetime(2026, 1, 1), catchup=False) as d:
    PythonOperator(task_id="load", python_callable=lambda: None,
                   outlets=[CUSTOMERS_RAW])

# Team analytics — consumers of ingest, producers of dimensional
with DAG("build_fact_orders_dag",
         schedule=[ORDERS_RAW],
         start_date=datetime(2026, 1, 1), catchup=False) as d:
    PythonOperator(task_id="build", python_callable=lambda: None,
                   outlets=[FACT_ORDERS])

with DAG("build_dim_customer_dag",
         schedule=[CUSTOMERS_RAW],
         start_date=datetime(2026, 1, 1), catchup=False) as d:
    PythonOperator(task_id="build", python_callable=lambda: None,
                   outlets=[DIM_CUSTOMER])

# Team reporting — consumer of dimensional, producer of final report
with DAG("daily_revenue_dag",
         schedule=[FACT_ORDERS, DIM_CUSTOMER],
         start_date=datetime(2026, 1, 1), catchup=False) as d:
    PythonOperator(task_id="build_revenue", python_callable=lambda: None,
                   outlets=[DAILY_REVENUE])

Step-by-step explanation.

1. The URI namespace uses three schemes: s3:// for raw ingest (parquet on S3), snowflake:// for dimensional (tables in Snowflake), and snowflake://reporting/... for final reports. The scheme carries semantic meaning; the catalog view can filter on it.
2. Team ingest emits two raw datasets on cron schedules (source system reads are cron-anchored). Team analytics consumes on dataset triggers; team reporting consumes downstream analytics datasets on AND semantics.
3. The Datasets tab shows five URIs. Each row displays the producer DAG, the consumer DAG(s), and the recent event history. A search for "orders" turns up s3://ingest/orders (producer: ingest_orders_dag; consumers: build_fact_orders_dag) and snowflake://analytics/fact_orders (producer: build_fact_orders_dag; consumers: daily_revenue_dag).
4. The lineage graph derived from this: ingest_orders_dag → s3://ingest/orders → build_fact_orders_dag → snowflake://analytics/fact_orders → daily_revenue_dag → snowflake://reporting/daily_revenue. Three hops, three teams, zero cross-team polling glue.
5. When a new team (say, team-ml) wants to build a model on orders, they browse the tab, find s3://ingest/orders, and write a new consumer DAG train_orders_model_dag with schedule=[ORDERS_RAW]. No coordination with team ingest is required. Team ingest doesn't even know team-ml exists — but the tab does.

Output.

URI	Producer	Consumers	Recent event	Downstream count
s3://ingest/orders	ingest_orders_dag	build_fact_orders_dag	05:07 today	1
s3://ingest/customers	ingest_customers_dag	build_dim_customer_dag	05:12 today	1
snowflake://analytics/fact_orders	build_fact_orders_dag	daily_revenue_dag	05:20 today	1
snowflake://analytics/dim_customer	build_dim_customer_dag	daily_revenue_dag	05:25 today	1
snowflake://reporting/daily_revenue	daily_revenue_dag	none (yet)	05:35 today	0

Rule of thumb. Design the URI catalog like an internal API surface. The naming convention, the schema-registry pairing, the ownership boundaries — these all matter more than the specific storage backing each URI. The catalog view is the payoff.

Worked example — versioned URI migration

Detailed explanation. Team ingest needs to change the schema of s3://ingest/orders — a breaking change (rename a column, drop a field). Cutting over in place would break every consumer. The canonical pattern: dual-emit the new URI s3://ingest/orders_v2 alongside the old for a deprecation window; consumers migrate one at a time; when the last consumer has cut over, delete the old outlet.

• Current. s3://ingest/orders — old schema.
• Target. s3://ingest/orders_v2 — new schema.
• Deprecation window. 30 days of dual-emit.
• Cutover. Consumers migrate; then s3://ingest/orders is retired.

Question. Write the producer during the dual-emit phase and show the consumer migration.

Input.

Phase	Producer emits	Consumers on	Consumers on
Before	orders (old)	orders (old)	—
Dual-emit	orders + orders_v2	mix of old / new	—
After	orders_v2	orders_v2	—

Code.

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

# During dual-emit phase — both URIs are outlets on the producer
ORDERS_OLD = Dataset("s3://ingest/orders")
ORDERS_V2  = Dataset("s3://ingest/orders_v2")

with DAG(
    dag_id="ingest_orders_dag",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as producer:

    load_old = PythonOperator(
        task_id="load_old_schema",
        python_callable=lambda: None,
        outlets=[ORDERS_OLD],           # emits old-schema event
    )

    load_new = PythonOperator(
        task_id="load_new_schema",
        python_callable=lambda: None,
        outlets=[ORDERS_V2],            # emits new-schema event
    )
    # Both tasks run in parallel; either can succeed independently.

# Consumer during migration — pin to whichever URI matches your driver code
with DAG(
    dag_id="analytics_new_consumer_dag",
    schedule=[ORDERS_V2],               # migrated to v2
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as consumer_v2:
    PythonOperator(task_id="do", python_callable=lambda: None)

# Legacy consumer during migration — still on old URI
with DAG(
    dag_id="legacy_consumer_dag",
    schedule=[ORDERS_OLD],              # not yet migrated
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as consumer_old:
    PythonOperator(task_id="do", python_callable=lambda: None)

Step-by-step explanation.

1. During the dual-emit phase, the producer DAG has two load tasks — one writing the old schema and emitting the old URI, one writing the new schema and emitting the new URI. Both tasks run in parallel; failures are independent.
2. The tab shows both URIs. Each has one producer (the same DAG) and its own set of consumers. Team-analytics can grep for "orders" and see both entries with a clear "v2" marker.
3. Consumers migrate one at a time. analytics_new_consumer_dag is on v2 (new-schema code inside). legacy_consumer_dag is still on the old URI. Both work in parallel; neither blocks the other.
4. When all consumers have migrated (verified by grepping the codebase for Dataset("s3://ingest/orders") and finding zero references outside of legacy_consumer_dag), the platform team pings the last consumer team and completes the migration.
5. Cleanup: delete the load_old_schema task from the producer; delete the ORDERS_OLD dataset reference; the URI disappears from the tab after the last event ages out. The producer is now single-emit on v2.

Output.

Time	Consumer_old state	Consumer_v2 state	Both URIs firing?
T0 (pre-migration)	active on old	doesn't exist	old only
T1 (dual-emit start)	active on old	active on v2	both
T2 (migration in progress)	migrating	active on v2	both
T3 (last consumer migrated)	deleted	active on v2	both (one week grace)
T4 (producer cleanup)	—	active on v2	v2 only

Rule of thumb. Treat URI renames like database schema migrations — dual-emit, deprecation window, verify zero references, cut over. Never rename a URI in place; the resulting outage will page the entire on-call rotation.

Worked example — freshness SLA per URI

Detailed explanation. A URI is a promise — team ingest commits to emit s3://ingest/orders at least once every 24 hours. The freshness SLA is the observable form of that promise. When the URI hasn't emitted an event in 24 hours, an alert fires and paging responsibility falls to the producer team. This is the operational contract that makes cross-team datasets work at scale.

• SLA. s3://ingest/orders must have a dataset event within the last 24 hours.
• Monitoring. A cron-scheduled check_freshness_dag queries the Airflow metadata DB for the latest event per URI.
• Alert. Slack + PagerDuty to the producing team when SLA violated.

Question. Write the freshness-check DAG that queries Airflow metadata for the latest dataset event and alerts on staleness.

Input.

URI	SLA (hours)	Owner
s3://ingest/orders	24	team-ingest
s3://ingest/customers	24	team-ingest
snowflake://analytics/fact_orders	26 (24 + 2 processing)	team-analytics
snowflake://analytics/dim_customer	26	team-analytics

Code.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models.dataset import DatasetModel
from airflow.utils.session import provide_session
from datetime import datetime, timedelta, timezone

SLA_HOURS = {
    "s3://ingest/orders":                        24,
    "s3://ingest/customers":                     24,
    "snowflake://analytics/fact_orders":         26,
    "snowflake://analytics/dim_customer":        26,
}

@provide_session
def _check_freshness(session=None, **ctx):
    """Query Airflow metadata for the latest event per URI, alert on staleness."""
    now = datetime.now(timezone.utc)
    stale = []
    for uri, sla_hours in SLA_HOURS.items():
        dataset = session.query(DatasetModel).filter_by(uri=uri).first()
        if not dataset:
            stale.append((uri, "no such dataset registered"))
            continue
        latest_event = max(
            (evt.timestamp for evt in dataset.consuming_dags[0].dataset_events),
            default=None,
        ) if dataset.consuming_dags else None
        if latest_event is None:
            stale.append((uri, "no events ever"))
            continue
        age_hours = (now - latest_event).total_seconds() / 3600.0
        if age_hours > sla_hours:
            stale.append((uri, f"stale by {age_hours - sla_hours:.1f}h"))

    if stale:
        _page_oncall(stale)
    else:
        print("all URIs fresh")

def _page_oncall(stale):
    for uri, reason in stale:
        print(f"ALERT: {uri} — {reason}")
    # In production: post to Slack + PagerDuty; include URI owner from a static map

with DAG(
    dag_id="check_dataset_freshness_dag",
    schedule="*/15 * * * *",             # every 15 minutes
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["platform", "freshness"],
) as dag:

    check = PythonOperator(
        task_id="check_freshness",
        python_callable=_check_freshness,
    )

Step-by-step explanation.

1. The DAG runs every 15 minutes on a cron schedule. It's a platform-owned DAG (not team-owned) that monitors freshness for every URI in the SLA_HOURS dict. This dict is the durable operational contract; changes require review.
2. For each URI, the callable queries the Airflow metadata database's DatasetModel to find the latest dataset event timestamp. The query uses Airflow's internal ORM — direct DB access is legitimate because this is platform tooling.
3. The age check is (now - latest_event) > sla_hours. If true, the URI is stale; it's added to the stale list with a reason. Reasons distinguish "no such dataset" (typo or new URI never emitted) from "no events ever" (registered but never fired) from "stale by Xh" (fired but not recently enough).
4. The _page_oncall function is the alerting side. In production it posts to Slack channels owned by the producing team and, for critical URIs, escalates to PagerDuty. A URI-owner map (from ADRs) maps URI to team.
5. Freshness alerts are the operational teeth of the URI contract. Without them, a producer team can silently stop emitting a URI and nobody notices until downstream reports go blank. With freshness alerts, the producer team is paged within minutes of an SLA violation.

Output.

URI	Latest event	Age	SLA	Status
s3://ingest/orders	06:07 today	8h	24h	fresh
s3://ingest/customers	06:12 today	8h	24h	fresh
snowflake://analytics/fact_orders	06:30 today	8h	26h	fresh
snowflake://analytics/dim_customer	26h ago	26h	26h	STALE — page team-analytics

Rule of thumb. Every URI ships with a freshness SLA and an alert. Without the SLA + alert pair, the URI contract is unenforceable and cross-team datasets degrade into a silent-failure surface. The freshness-check DAG is 30 lines of platform tooling; not writing it is a common pre-incident mistake.

Senior interview question on the datasets marketplace pattern

A senior interviewer might ask: "You're the platform lead for an Airflow deployment growing from 3 teams to 12. Walk me through how you'd stand up the datasets tab as a marketplace — URI naming conventions, ownership model, freshness SLAs, discoverability, and the runbook for a URI-related incident."

Solution Using a URI-marketplace platform pattern

# 1. URI naming ADR — locked in the platform wiki
"""
Convention: <storage_scheme>://<owning_team>/<logical_name>[_v<major>]

Storage schemes (locked list):
    s3://       parquet / json / csv on S3
    gs://       parquet on Google Cloud Storage
    azure://    blob storage
    snowflake://   Snowflake tables
    bigquery://    BigQuery tables
    logical://   synthetic / event-only, no physical storage
    custom://    other; must be documented in the ADR

Owning team is one of the platform's registered team slugs.
Logical name is snake_case, plural for tables (orders), singular for singletons (gl_close_status).
Version suffix _v2 only on breaking changes.

Forbidden in URIs:
    - environment prefixes (dev/stg/prd)
    - partition suffixes (dt=..., hr=...)
    - team-internal task or DAG names
"""

# 2. URI catalog exports — a nightly job dumps the tab to JSON for a searchable frontend
from airflow.decorators import dag, task
from airflow.models.dataset import DatasetModel
from airflow.utils.session import provide_session
from datetime import datetime

@dag(dag_id="platform_export_dataset_catalog_dag",
     schedule="0 3 * * *",
     start_date=datetime(2026, 1, 1),
     catchup=False,
     tags=["platform"])
def export_catalog():

    @task
    @provide_session
    def dump(session=None):
        rows = []
        for ds in session.query(DatasetModel).all():
            rows.append({
                "uri":       ds.uri,
                "producers": [d.dag_id for d in ds.producing_tasks],
                "consumers": [d.dag_id for d in ds.consuming_dags],
                "latest":    ds.latest_event_timestamp,
            })
        _publish_catalog(rows)   # POST to internal search UI

    dump()

export_catalog()

# 3. Freshness SLA registry — one YAML per team, merged at platform level
"""
# team-ingest/freshness.yaml
- uri: s3://ingest/orders
  sla_hours: 24
  owners: [ingest-oncall]
  criticality: high
- uri: s3://ingest/customers
  sla_hours: 24
  owners: [ingest-oncall]
  criticality: high
"""

# 4. URI incident runbook (excerpted)
"""
Stage 1 — Confirm: Datasets tab shows URI last event > SLA hours ago
Stage 2 — Route:   Look up owner from freshness.yaml; page owning team
Stage 3 — Diagnose: Producer DAG state; task-level failure logs
Stage 4 — Mitigate: Fix producer OR manually inject event (Datasets tab affordance)
Stage 5 — Post-mortem: Update ADR if URI semantics need clarification
"""

Step-by-step trace.

Element	Purpose	Owner
URI naming ADR	canonical namespace	platform
Catalog export job	searchable frontend	platform
Freshness YAML per team	SLA registry	producing team
Freshness alert DAG	SLA enforcement	platform
Datasets tab	native discoverability	Airflow
URI incident runbook	ops playbook	platform

After the marketplace pattern is stood up, adding a new team is a 30-minute onboarding: register the team slug, add their freshness.yaml, teach their engineers to browse the Datasets tab, and ship. The URI catalog becomes the platform's default cross-team data contract; Slack threads about "who owns X" evaporate.

Output:

Growth stage	URI count	Cross-team overhead
3 teams	~15 URIs	none
6 teams	~40 URIs	catalog search wins
12 teams	~120 URIs	catalog search is critical

Why this works — concept by concept:

• URI ADR as governance — locking the naming convention in an ADR turns URI design from ad-hoc into a reviewable decision. Every new URI is a one-line ADR update, not a Slack debate.
• Catalog export for searchability — the Airflow Datasets tab is discoverable within Airflow; exporting to an internal search frontend (Backstage, custom UI) makes it discoverable across the whole engineering org.
• Freshness YAML per team — decentralised registry (team owns their YAML), centralised alerting (platform runs the checker). Ownership stays with the producer; enforcement stays with the platform.
• Runbook + tab affordances — the tab supports manual event injection for one-off mitigation. The runbook uses this affordance during real incidents while the underlying producer bug is fixed.
• Cost — the platform pattern is O(1) engineering time to stand up (a week); the per-team onboarding is O(1) time (30 minutes). The marketplace becomes cheaper as teams grow, not more expensive. Compare to the pre-datasets world where each new team meant N new cross-DAG sensors.

ETL
Topic — etl
ETL problems on cross-team data contracts

Practice →

SQL Topic — sql SQL problems on lineage and downstream propagation

Practice →

---

5. Airflow 3.x additions + production migration

Filters, custom triggers, freshness monitors — the 3.x additions that make datasets a mature primitive

The mental model in one line: Airflow 3.x adds three things on top of the 2.4 datasets primitive — payload-aware filters (fire the consumer only when an event predicate is true), custom triggers on datasets (arbitrary Python that decides whether an event fires downstream), and richer event metadata — plus the mature migration recipe for retiring the last ExternalTaskSensor in your codebase. Every 3.x-era interview probe on datasets assumes you know both the 2.4 baseline and the 3.x extensions.

The five 3.x additions worth naming.

• Dataset events with payload. In 3.x, dataset events carry a JSON payload — arbitrary metadata about the run (row count, partition, source snapshot ID). The scheduler passes the payload to consumer DAGs as a run-context field.
• Dataset event filters. Consumers can declare a predicate on the event payload: schedule=DatasetTriggeredTimetable(datasets=[...], filter=lambda evt: evt.extra["row_count"] > 0). Only events passing the predicate fire the consumer.
• Custom triggers on datasets. Beyond filters, custom trigger classes can implement arbitrary trigger logic — including composition across multiple dataset events (OR semantics, time windows, complex predicates).
• Consumer-side event history. Consumers can query the recent event history of their consumed datasets via a run-context helper, useful for late-arriving-data handling.
• Dataset aliases. A single logical dataset can be referenced by multiple aliases (s3://ingest/orders == orders), simplifying URI migration and shortening consumer code.

The dataset filter pattern.

• Purpose. Consume only events that pass a predicate. E.g., fire the consumer only when the row count is nonzero, or only when a source-partition marker is set.
• Placement. On the consumer's schedule declaration. The scheduler evaluates the predicate at event-arrival time; failed predicates are silently dropped.
• Determinism. The predicate must be pure — no side effects, no external calls, no time-dependent logic. The scheduler evaluates it eagerly; non-determinism causes replay inconsistencies.
• Backward compatibility. Filters are 3.x only; 2.4 → 2.7 consumers must move the predicate check into the consumer DAG's first task and short-circuit if false.

The custom trigger class pattern.

• Purpose. Trigger logic that filters cannot express — OR across datasets, time-windowed AND, "fire on the third event in the last hour," etc.
• Shape. Subclass BaseTrigger (or the dataset-specific trigger class), implement should_fire(event, state), register on the consumer's schedule.
• Use cases. Bursty ingest where you want to batch multiple events; ML retraining that fires on either a feature-store update or a labelled-data update (OR); rate-limited consumers that fire at most once per hour regardless of event frequency.
• Cost. Custom trigger code is trickier than a simple filter; unit-test heavily.

The ExternalTaskSensor migration recipe.

• Step 1 — inventory. grep -R ExternalTaskSensor dags/ to enumerate every occurrence. Categorise by source DAG.
• Step 2 — add outlets. For each source DAG named in a sensor, add outlets=[Dataset(URI)] on the source's final task. Choose URI names by convention. Ship without touching the consumer.
• Step 3 — replace consumer. Rewrite the consumer's schedule from schedule="..." + ExternalTaskSensor + downstream_tasks to schedule=[Dataset(URI)] + downstream_tasks. Delete the sensor.
• Step 4 — parallel-run week. Keep the old consumer running with a _legacy suffix on the DAG ID; watch for output divergence.
• Step 5 — delete legacy. After a week of matching outputs, delete the legacy consumer. Done.

The freshness monitor as production discipline.

• Purpose. Every URI ships with an SLA; the platform enforces it. The absence of freshness monitoring makes datasets a silent-failure surface.
• Implementation. Cron-scheduled DAG queries the Airflow metadata DB for latest event per URI; compares to SLA table; pages on violation.
• Signal. The senior interview signal is naming freshness monitoring as a required companion to datasets, not an afterthought.

Common interview probes on 3.x + production.

• "What's new in Airflow 3.x datasets vs 2.4?" — payload filters, custom triggers, richer metadata.
• "How would you fire a consumer only if the producer wrote nonzero rows?" — 3.x filter predicate on row_count.
• "What's the migration recipe from ExternalTaskSensor?" — inventory, add outlets, replace consumer, parallel-run, delete.
• "How do you monitor dataset freshness?" — platform DAG queries metadata DB, alerts on SLA violation.
• "When would you write a custom trigger class?" — OR semantics, time-windowed logic, rate limiting.

Worked example — 3.x payload filter on row count

Detailed explanation. A producer emits s3://ingest/orders even on empty days (source system has no new orders on weekends). A consumer should fire only on nonzero-row days. Airflow 3.x filter predicates express this in a single lambda; the pre-3.x workaround was a first-task ShortCircuitOperator in the consumer.

• Producer. ingest_orders_dag, emits event with payload {"row_count": N}.
• Consumer. expensive_transform_dag, should fire only when row_count > 0.
• Filter. lambda evt: evt.extra["row_count"] > 0.

Question. Write the producer emitting the payload and the consumer with the filter.

Input.

Component	Value
Producer emit	event with extra={"row_count": n_rows}
Consumer filter	row_count > 0

Code.

# Producer — emit event with payload
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime

ORDERS = Dataset("s3://ingest/orders")

def _load_and_emit(**ctx):
    n_rows = _write_to_s3()
    # 3.x — attach payload to the event
    ctx["outlet_events"][ORDERS].extra = {"row_count": n_rows}
    print(f"emitted with row_count={n_rows}")

def _write_to_s3() -> int:
    # returns count of rows written (may be 0)
    return 12345

with DAG(
    dag_id="ingest_orders_dag",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as producer:
    PythonOperator(
        task_id="load",
        python_callable=_load_and_emit,
        outlets=[ORDERS],
    )


# Consumer — filter on payload
from airflow.timetables.datasets import DatasetTriggeredTimetable

def _row_count_positive(event) -> bool:
    return event.extra.get("row_count", 0) > 0

filtered_schedule = DatasetTriggeredTimetable(
    datasets=[ORDERS],
    event_filter=_row_count_positive,   # 3.x — event predicate
)

with DAG(
    dag_id="expensive_transform_dag",
    schedule=filtered_schedule,
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as consumer:
    PythonOperator(task_id="do", python_callable=lambda: None)

Step-by-step explanation.

1. The producer's task callable receives an outlet_events mapping in its context — the 3.x mechanism for attaching payload to the fired event. Setting ctx["outlet_events"][ORDERS].extra = {...} puts the payload on the emitted event.
2. The consumer declares its schedule as a DatasetTriggeredTimetable (the 3.x timetable class) with an event_filter callable. The scheduler evaluates the callable on each incoming event and drops events that return False.
3. On a normal weekday when the producer writes 12345 rows, the payload {"row_count": 12345} is attached, the filter returns True, the consumer fires.
4. On a weekend when the producer writes 0 rows, the payload is {"row_count": 0}, the filter returns False, the event is dropped. The consumer does not fire. No wasted compute.
5. The pre-3.x equivalent required a first-task ShortCircuitOperator in the consumer that checked S3 for a marker file or queried a metadata DB — significantly more scheduling glue. 3.x moves the check into the scheduler itself.

Output.

Producer run	Rows written	Payload	Filter result	Consumer fires?
Weekday	12345	{"row_count": 12345}	True	yes
Weekend	0	{"row_count": 0}	False	no
Holiday	42	{"row_count": 42}	True	yes

Rule of thumb. For "consume only under a predicate" logic, use 3.x filters if available. They express intent at the schedule layer, not inside consumer tasks, which is where the semantics belong. Filters must be pure and cheap — avoid I/O in the predicate.

Worked example — custom trigger class for OR semantics

Detailed explanation. A DAG that should fire when either dataset A or dataset B emits an event — the OR-semantics case that the base schedule=[A, B] primitive does not support. A custom trigger class subclasses the dataset-triggered timetable and overrides the fire condition.

• Consumer. ml_train_dag, fires on either features update or labels update.
• Datasets. s3://ml/features, s3://ml/labels.
• Trigger. Custom class implementing OR.

Question. Write the custom trigger class and the consumer that uses it.

Input.

Component	Value
Dataset A	s3://ml/features
Dataset B	s3://ml/labels
Semantics	OR — fire on either

Code.

# custom_or_trigger.py — subclass DatasetTriggeredTimetable
from airflow.timetables.datasets import DatasetTriggeredTimetable
from typing import List

class DatasetOrTriggered(DatasetTriggeredTimetable):
    """Fire on any single event from any dataset in the list (OR semantics)."""

    def next_dagrun_info(self, last_automated_data_interval, restriction):
        # Fire on any single dataset event since the last consumer run.
        # This overrides the base AND semantics.
        events = self._collect_new_events(last_automated_data_interval)
        if events:
            return self._make_dagrun_info(events[0].timestamp)
        return None


# consumer_dag.py
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime
from custom_or_trigger import DatasetOrTriggered

FEATURES = Dataset("s3://ml/features")
LABELS   = Dataset("s3://ml/labels")

or_schedule = DatasetOrTriggered(datasets=[FEATURES, LABELS])

with DAG(
    dag_id="ml_train_dag",
    schedule=or_schedule,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["ml", "train"],
) as dag:
    PythonOperator(task_id="train", python_callable=lambda: None)

Step-by-step explanation.

1. The DatasetOrTriggered class subclasses the base timetable and overrides next_dagrun_info. The override collects new events across all listed datasets and fires if any event arrived since the last run — OR semantics.
2. The consumer declares schedule=or_schedule using an instance of the custom class. The rest of the consumer DAG is unchanged; the trigger semantics live entirely in the timetable.
3. When features update at 05:00, the OR condition is satisfied; the consumer fires. The labels dataset is unchanged; that's fine.
4. When labels update at 06:00, the OR is satisfied again; the consumer fires again. Each dataset event fires the consumer independently.
5. The tradeoff of OR vs AND: OR fires more often (potentially wasteful compute); AND fires only when quorum is reached (potentially stale). Choose based on the workload; the custom-trigger escape hatch means you can always express the right semantics.

Output.

Event	Time	OR condition	Consumer fires?
features updates	05:00	true (features new)	yes
labels updates	06:00	true (labels new)	yes
both update	07:00	true	yes (once)
neither updates	08:00	false	no

Rule of thumb. Custom trigger classes are the escape hatch for semantics the base primitive cannot express — OR, time-windowed AND, rate limiting. Write one class, test it, ship it in a platform module, reuse across consumers. Custom triggers should be a small library, not a per-DAG special case.

Worked example — end-to-end ExternalTaskSensor → dataset migration

Detailed explanation. The migration playbook applied to a concrete DAG: transform_orders_dag currently opens with an ExternalTaskSensor pointing at ingest_orders_dag's load_to_s3 task. Rewrite as datasets; run in parallel for a week; retire the legacy.

• Legacy consumer. cron 05:30 + ExternalTaskSensor(external_dag_id="ingest_orders_dag", external_task_id="load_to_s3", execution_delta=timedelta(minutes=30)).
• New consumer. schedule=[Dataset("s3://ingest/orders")], no sensor.
• Producer. outlets=[Dataset("s3://ingest/orders")] on load_to_s3.

Question. Show all three states — before, dual-run, after — and the smoke-test that confirms output parity during the dual-run week.

Input.

State	Producer	Legacy consumer	New consumer
Before	no outlets	cron + sensor	doesn't exist
Dual-run	outlets added	cron + sensor	dataset schedule
After	outlets	deleted	dataset schedule

Code.

# Before — legacy consumer with sensor
from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

with DAG(
    dag_id="transform_orders_dag",       # legacy
    schedule="30 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as legacy:
    wait = ExternalTaskSensor(
        task_id="wait",
        external_dag_id="ingest_orders_dag",
        external_task_id="load_to_s3",
        execution_delta=timedelta(minutes=30),
        poke_interval=60,
        mode="reschedule",
    )
    do = PythonOperator(task_id="do", python_callable=lambda: None)
    wait >> do


# Step 1 — add outlet to producer (no consumer change yet)
from airflow import Dataset

ORDERS = Dataset("s3://ingest/orders")

with DAG(dag_id="ingest_orders_dag", schedule="0 5 * * *",
         start_date=datetime(2026, 1, 1), catchup=False) as producer:
    PythonOperator(task_id="load_to_s3", python_callable=lambda: None,
                   outlets=[ORDERS])


# Step 2 — new consumer running in parallel (dual-run week)
with DAG(
    dag_id="transform_orders_v2_dag",     # new; parallel to legacy
    schedule=[ORDERS],
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as new_consumer:
    do = PythonOperator(task_id="do", python_callable=lambda: None)


# Step 3 — after a week of parity, rename new to canonical name; delete legacy
# Rename transform_orders_v2_dag → transform_orders_dag
# Delete the legacy DAG file

Step-by-step explanation.

1. Step 1 adds the outlet to the producer. This is a pure additive change — the legacy sensor keeps working; the new outlet fires an event that nothing yet consumes. Ship this change and watch the Datasets tab confirm the URI appears.
2. Step 2 stands up the new consumer under a temporary DAG ID (_v2 suffix). It has the same task graph as the legacy but a dataset-triggered schedule instead of cron + sensor. Both DAGs run every day; both do the same work.
3. Dual-run parity check: at the end of each day, compare the output of the legacy DAG's do task with the new consumer's do task. If both produce identical outputs (same row counts, same aggregates, same file hashes), parity is confirmed. Any divergence indicates a schedule-timing difference — investigate before proceeding.
4. After a week of clean parity, cut over: rename transform_orders_v2_dag → transform_orders_dag (delete the legacy first, then rename to reclaim the DAG ID). The dataset-triggered version is now the canonical DAG.
5. Cleanup: delete the legacy DAG file entirely; delete the ExternalTaskSensor import from your DAGs directory (a lint rule to reject future imports is a nice touch). The migration is complete.

Output.

Day	Legacy runs at	New consumer runs at	Output parity
Day 1	05:30	05:07 (dataset event)	matches
Day 2	05:30	05:12 (producer slipped 5m)	matches
Day 3	05:30	05:05	matches
Day 4	05:30	05:15	matches
Day 5	05:30	05:09	matches
Day 6	(no producer event; sensor waits then times out)	(no event; new consumer stays quiet)	matches (both did nothing)
Day 7	05:30	05:08	matches → cut over

Rule of thumb. Every ExternalTaskSensor migration follows the same three steps: producer outlet, parallel new consumer, dual-run week, cut over. The parity check during the dual-run week is the safety net — trust it, don't skip it.

Senior interview question on production migration to datasets

A senior interviewer might ask: "You're leading the migration of an Airflow 2.3 deployment (pre-datasets) with 80 DAGs and 40 ExternalTaskSensor edges. You've upgraded to Airflow 3.x. Walk me through the full migration plan — sequencing, tooling, safety nets, and what you'd instrument to know you're done."

Solution Using a phased, tooled, monitored migration

# Migration tooling — inventory + tracker
"""
Migration inventory YAML — one row per sensor edge
=================================================
- consumer_dag: transform_orders_dag
  legacy_sensor:
    external_dag_id: ingest_orders_dag
    external_task_id: load_to_s3
  target_uri: s3://ingest/orders
  producer_dag: ingest_orders_dag
  producer_task: load_to_s3
  status: planned      # planned -> outlet_added -> dual_running -> cut_over -> done
  parity_days: 0

- consumer_dag: transform_customers_dag
  ...
"""

# Phase 1 — automated inventory
import ast, pathlib

def inventory_sensors(dags_dir: pathlib.Path):
    edges = []
    for path in dags_dir.rglob("*.py"):
        tree = ast.parse(path.read_text())
        for node in ast.walk(tree):
            if isinstance(node, ast.Call) and getattr(node.func, "attr", "") == "ExternalTaskSensor":
                edges.append({
                    "file": str(path),
                    "line": node.lineno,
                })
    return edges

# Phase 2 — add outlets in a batch PR (all producers at once)
# Phase 3 — flip one consumer per day, run in parallel, verify parity
# Phase 4 — bulk cleanup after all consumers migrated

# Phase 5 — CI lint rule to prevent regression
"""
# .pre-commit-hooks.yaml
- id: no-external-task-sensor
  entry: rg -q "ExternalTaskSensor" dags/ && echo "ExternalTaskSensor forbidden; use datasets"
  language: system
"""

# Phase 6 — freshness dashboard
from airflow.decorators import dag, task
from airflow.models.dataset import DatasetModel
from datetime import datetime, timezone

@dag(dag_id="migration_progress_dag",
     schedule="0 * * * *",
     start_date=datetime(2026, 1, 1),
     catchup=False)
def track_progress():

    @task
    def report():
        # Count of URIs with recent events → progress metric
        # Count of remaining ExternalTaskSensor imports (from CI export) → regression metric
        _publish_dashboard()

    report()

Step-by-step trace.

Phase	Weeks	Deliverable
1 — Inventory	1	migration.yaml with 40 edges
2 — Add outlets (batch PR)	2	40 producer DAGs carry outlets; sensors untouched
3 — Flip consumers (one per day)	3–10	40 consumers on datasets; legacy kept in parallel
4 — Parity monitoring	4–11	Daily parity report; divergences investigated
5 — Bulk legacy cleanup	12	Sensor DAGs deleted; imports removed
6 — CI lint	12	Pre-commit rule prevents new ExternalTaskSensor
7 — Freshness alerting	13	Every URI has an SLA; platform monitors

At week 13, the deployment is entirely on datasets. Zero sensors; zero cron offsets between related DAGs; a URI marketplace in the Datasets tab; a freshness SLA per URI with paging on violation; a lint rule preventing regression. The senior signal is the completeness of the plan — inventory + rollout + parity + cleanup + regression guard + freshness.

Output:

Metric	Before	After
ExternalTaskSensor edges	40	0
TriggerDagRunOperator edges	~10	0
Hand-tuned cron offsets	~40	0
URIs in Datasets tab	0	~35
Freshness SLAs defined	0	35
Migration weeks	—	13
CI regression guard	none	pre-commit rule

Why this works — concept by concept:

• Inventory-driven rollout — a machine-readable inventory YAML lets the team track progress row-by-row. No sensor is forgotten; no consumer is orphaned. The YAML doubles as the durable migration record.
• Additive-first sequencing — adding outlets before touching consumers is pure additive; the legacy sensors keep working while producers get datasets. Zero-risk phase 2.
• Parity checks as safety net — daily parity checks during the dual-run week catch schedule-timing differences before they cause data-quality incidents. The parity YAML is the safety switch.
• CI lint against regression — after cutover, a pre-commit lint rejects any future ExternalTaskSensor import. Prevents the deployment from silently accumulating new sensors as teams onboard.
• Freshness monitor as the maturity signal — the last step is not "migration done" but "freshness SLA per URI." Datasets without freshness monitoring are a silent-failure surface; datasets with freshness monitoring are a mature production primitive. The senior signal is naming this as the completion criterion.
• Cost — engineering cost is O(sensors) once during the phased rollout; operational cost is O(1) afterward. Every future cross-DAG dependency becomes a single-line URI declaration on both sides instead of a bespoke sensor. The avoided cost of a single missed-cutover incident typically pays back the migration effort in the first month.

ETL
Topic — etl
ETL problems on Airflow migration and datasets

Practice →

Optimization
Topic — optimization
Optimization problems on freshness and event-driven pipelines

Practice →

---

Cheat sheet — Airflow datasets recipes

• Producer outlets 3-line template. On the durable-write task, add outlets=[Dataset("s3://<team>/<name>")]. That's the entire producer-side change. Success on that task emits one event; failure emits none; retries collapse to one event. Never place outlets on intermediate tasks; always on the final write.
• Consumer schedule=[Dataset(...)] one-liner. Replace schedule="0 5 * * *" with schedule=[Dataset("s3://<team>/<name>")] and delete every ExternalTaskSensor pointing at the same producer. Multiple datasets in the list are AND semantics — the consumer fires when all have new events since the last run.
• Cron + dataset combo (DatasetOrTimeSchedule). For "fire when data arrives OR at a cron backstop," wrap with DatasetOrTimeSchedule(timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"), datasets=[Dataset(URI)]). Design the consumer to be idempotent — the compound schedule can double-fire on late-arriving events.
• Dataset URI naming convention. <storage_scheme>://<owning_team>/<logical_name>[_v<major>]. Storage schemes are s3://, snowflake://, bigquery://, logical://, custom://. Forbidden in URIs: environment prefixes (dev/stg/prd), partition suffixes (dt=...), team-internal task names. Version suffix only on breaking changes.
• ExternalTaskSensor migration recipe. (1) Inventory every sensor into a YAML. (2) Add outlets=[Dataset(URI)] to the producer's final task (additive, zero risk). (3) Rewrite each consumer with schedule=[Dataset(URI)] under a _v2 DAG ID. (4) Run legacy + new in parallel for a week; verify output parity daily. (5) Cut over, delete legacy, add a CI lint rejecting new ExternalTaskSensor imports.
• Producer success semantics. The scheduler emits a dataset event exactly when the outlet-bearing task reaches success. failed, upstream_failed, skipped, canceled states emit no event. Retries that eventually succeed emit exactly one event. Success on zero-row writes still emits — guard with an explicit if n_rows == 0: raise ....
• Multi-dataset AND across producers. schedule=[A, B, C] fires only when all three have events since the last consumer run. The consumer fires at the moment the last quorum event arrives — bounded by the slowest producer.
• Airflow 3.x payload filter. Attach {"row_count": n} to the event via ctx["outlet_events"][DS].extra; declare DatasetTriggeredTimetable(datasets=[DS], event_filter=lambda evt: evt.extra["row_count"] > 0) on the consumer. Predicate must be pure and cheap.
• Custom trigger class for OR / windowed logic. Subclass DatasetTriggeredTimetable, override next_dagrun_info, register on the consumer's schedule. Escape hatch for semantics the base primitive can't express — OR across datasets, time-windowed AND, rate limiting.
• Datasets tab as marketplace. The Airflow UI's Datasets tab lists every URI, its producer DAG(s), its consumer DAG(s), and recent events. Grep-searchable. Manual event injection affordance for one-off downstream triggers.
• Freshness SLA per URI. Every URI ships with an SLA (e.g. 24h). A platform-owned cron DAG queries the Airflow metadata DatasetModel for latest event per URI; pages the producing team on SLA violation. Without freshness monitoring, datasets are a silent-failure surface.
• Outlets vs inlets. outlets on a task trigger downstream DAGs on success. inlets are lineage-only annotations — visible in the Datasets tab, but do not trigger anything. Never confuse the two; the interview probe is exactly this distinction.
• URI versioning for breaking changes. Never rename a URI in place. Dual-emit URI and URI_v2 for a deprecation window; consumers migrate one at a time; retire the old URI when the last consumer cuts over. Treat URI renames like database schema migrations.
• TaskFlow API works identically. @task(outlets=[Dataset(URI)]) is the TaskFlow syntax; it behaves identically to PythonOperator(outlets=[...]). Outlets are a scheduler-layer primitive, not an API-specific feature.
• Metadata-only URIs (logical:// or event://). For synthetic events with no physical storage (manual triggers, cross-team signals, orchestration flags), use a stable custom scheme. The scheduler treats the URI as opaque; the scheme carries semantics for humans reading the catalog.

Frequently asked questions

What are Airflow datasets and why did they replace cron-plus-ExternalTaskSensor cross-DAG orchestration?

Airflow datasets are logical scheduling objects introduced in Airflow 2.4 (September 2022) and matured through 2.7 and 3.x. A dataset is identified by a URI string (s3://team/name, snowflake://team/table, custom://team/signal) and represents a contract between producer and consumer DAGs — the producer declares outlets=[Dataset(URI)] on its durable-write task, the consumer declares schedule=[Dataset(URI)], and the scheduler wires the trigger. They replaced cron-plus-ExternalTaskSensor because the pre-2.4 model coupled consumers to producer internals (specific dag_id and task_id), required hand-tuned cron offsets that broke when producers slipped, held worker slots for polling, and made cross-team ownership fragile. Datasets collapse all of that into a URI contract: producers commit to emit; consumers subscribe; the scheduler handles routing, retries, and event history without either team writing scheduling glue.

Is the dataset URI the actual S3 or Snowflake path — or just a logical identifier?

Just a logical identifier. The Airflow scheduler treats the URI as an opaque event key — it never reads the URI, never checks whether the path exists, and never validates the schema underneath. s3://ingest/orders is a string; the scheduler cares only that the string is stable and that outlets/schedules pin to matching strings. This is deliberate: it lets you use custom:// or logical:// URIs for synthetic events (manual triggers, cross-team signals) that have no physical storage. The convention <storage_scheme>://<owning_team>/<logical_name> is human-readable and encodes ownership, but the storage scheme is metadata for humans, not a hint to the scheduler. The URI is a promise; the storage backing it is an implementation detail.

Can I combine dataset triggers with a cron schedule on the same DAG?

Yes — via DatasetOrTimeSchedule. Passing a plain list of datasets to schedule= gives pure event-driven behaviour with AND semantics across datasets. To combine "fire on any dataset event OR on a cron backstop," wrap the two in DatasetOrTimeSchedule(timetable=CronTriggerTimetable("0 7 * * *", timezone="UTC"), datasets=[Dataset(URI)]). The consumer fires whenever either trigger fires; both can fire on the same day (dataset event at 05:07, cron backstop at 07:00). Design the consumer to be idempotent — merge/upsert targets rather than append — because the compound schedule can produce duplicate runs on late-arriving events. The pattern is standard for compliance workloads where you must ship a report every day but prefer fresh data when available.

Does schedule=[Dataset(A), Dataset(B)] mean OR or AND — and how do consumer runs count events?

AND — the consumer fires only when all datasets in the list have received at least one event since the last consumer run. This is non-configurable in the base primitive. The scheduler tracks per-(consumer_dag, dataset) "last consumed event timestamp" markers. Each dataset event advances that dataset's marker; the fire condition is met when every marker in the schedule list has advanced past the last consumer run's timestamp. Events collapse: if dataset A fires three times between two consumer runs, those three fire the consumer once (on the third), and the consumer sees the latest state of A when it runs. For OR semantics, either write two separate consumer DAGs (each with a single dataset) sharing task code, or (Airflow 3.x) write a custom DatasetTriggeredTimetable subclass overriding next_dagrun_info to implement OR.

How do I migrate an existing ExternalTaskSensor-heavy Airflow deployment to datasets without stopping the world?

Five-phase additive migration. Phase 1: inventory every sensor into a YAML tracking (consumer_dag, producer_dag, producer_task, target_uri). Phase 2: add outlets=[Dataset(URI)] to each producer's durable-write task — this is purely additive; the legacy sensors keep working, the new outlet fires an event that nothing yet consumes. Ship this as a batch PR with zero risk. Phase 3: rewrite each consumer under a _v2 DAG ID with schedule=[Dataset(URI)]; keep the legacy DAG running in parallel. Phase 4: for one week, compare outputs of legacy vs new nightly; investigate any parity divergence before continuing. Phase 5: rename _v2 to canonical, delete the legacy DAG, add a pre-commit lint rule rejecting future ExternalTaskSensor imports. Finish by shipping a platform-owned freshness-monitor DAG that pages producer teams on SLA violation — datasets without freshness monitoring are a silent-failure surface.

What did Airflow 3.x add on top of the 2.4 datasets primitive?

Three practical additions plus richer metadata. (1) Event payload — dataset events now carry a JSON extra dict set by the producer (ctx["outlet_events"][DS].extra = {"row_count": n}); the payload is passed to the consumer as run context. (2) Event filters — consumers can attach a predicate on the payload (event_filter=lambda evt: evt.extra["row_count"] > 0) and the scheduler drops events failing the predicate; useful for "consume only nonzero-row updates." (3) Custom trigger classes — subclass DatasetTriggeredTimetable to implement semantics the base primitive can't express (OR across datasets, time-windowed AND, rate limiting). Plus (4) dataset aliases so a single logical dataset can be referenced by multiple URIs, easing renames, and (5) consumer-side event history helpers so a consumer's first task can inspect recent producer events for late-arriving-data logic. The 2.4 primitive was the foothold; 3.x is the mature surface.

Practice on PipeCode

• Drill the ETL practice library → for the cross-DAG orchestration, event-driven pipeline, and producer/consumer sequencing problems senior interviewers use to probe Airflow depth.
• Rehearse SQL transformations on the SQL practice library → for the downstream reshape and dimensional-model queries that consumer DAGs actually run.
• Tune throughput and freshness with the optimization practice library → for the SLA-monitor, back-pressure, and multi-producer scheduling problems.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the datasets + URI-contract intuition against real graded inputs.

Lock in Airflow datasets muscle memory

Airflow docs explain the API. PipeCode drills explain the decision — when transaction-anchored outlets replace ExternalTaskSensor, when AND semantics beat cron offsets, when `DatasetOrTimeSchedule` earns its double-fire risk, and when the freshness SLA is the real production primitive. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the orchestration trade-offs senior data engineers actually face.

Practice ETL problems →
Practice optimization problems →