Scheduling: Advanced

An Asset is a name and a URI standing for data that Airflow never opens. Declaring one changes nothing in storage; it gives 2 DAGs a shared word for the same table or file. The task that updates the data lists the asset in outlets, and a DAG that should run after each update puts the same asset in schedule, in place of a cron string or a timetable.

When land reaches success, Airflow writes an asset event: the asset, the instant, and the producing DAG, task and run. On its next loop the scheduler reads the new events, finds that orders_summary is scheduled on orders_raw, and creates a run of it, usually within seconds of the event. Nothing in this path reads a clock, and nothing inspects storage: the event means that a task declaring the outlet succeeded. A task that declares the outlet and writes nothing still fires it, and a task that writes the file and then fails fires nothing.

A scheduled run names a window, and its tasks read that window. An asset-triggered run has no window to name, because it exists for an update, and an update has no start and end of its own. Its logical_date is None, its data_interval_start and data_interval_end are None, and its run id starts with asset_triggered__ followed by the instant Airflow created it. A task that reads data_interval_start to pick a partition gets None and must be written for that. What the run does carry is triggering_asset_events in the task context: for each asset in the schedule, the list of events that caused the run, each holding its timestamp, source_dag_id, source_run_id and source_task_id. A consumer that needs the producer's window follows source_run_id back to the producing run and reads the window from there.

Events are kept until a run consumes them. If orders_summary is paused while the landing DAG succeeds 3 times, 3 events wait in its queue; on unpause the scheduler creates 1 run, and triggering_asset_events in that run lists all 3. The same batching happens whenever several events arrive between 2 scheduler loops. A consumer written to process everything since its last run handles this without noticing, and one written as if each run carried exactly 1 event reads the first and leaves 2 updates unprocessed.

The Assets page in the UI draws the graph the declarations imply: each asset in the middle, the DAGs whose tasks list it in outlets on one side, the DAGs scheduled on it on the other, and the time of its last event. In the grid of orders_summary, each run carries the asset-triggered marker in place of the clock, and the run's details list the events it consumed.

A producer whose whole job is 1 asset can be written with the @asset decorator from airflow.sdk, which builds a DAG of 1 task around the function and declares the asset as its outlet in the same step.

A daily revenue table needs orders and refunds together, and a fraud scorer needs either a new order batch or a new rule set. Asset schedules combine for both: & builds a schedule that waits for every asset in it, | builds one that fires on any of them, and parentheses nest the 2. Airflow keeps, per consumer DAG and per asset, a flag saying whether an event has arrived since that consumer last ran, and tests the expression against those flags on each scheduler loop.

daily_revenue needs 1 event from each of its 2 assets since its last run. Counts beyond 1 change nothing: 2 orders events and 0 refunds events leave it waiting. When the refunds event arrives, the run is created with all 3 events in triggering_asset_events, and both flags reset to waiting. fraud_scores fires on every single event of either asset, so 2 orders events 15 minutes apart give it 2 runs.

nightly_audit is satisfied by the pair or by a rules event alone. Over the same night it runs at 02:40 on the pair, at 03:10 on the rule set by itself, and at 04:05 on the pair again. Each run consumes every event queued for that DAG, whichever side of the expression matched. Had a rules event arrived at 03:55, after the 03:50 refunds event, it would have created a run carrying both and cleared the refunds flag, and the 04:05 orders event would then have left the pair waiting.

A report due at 06:00 whether the feed arrived mixes the 2 kinds of trigger. AssetOrTimeSchedule takes a timetable and an asset expression and creates a run when either side is satisfied, and the clock side fires every day whether or not the assets landed. With CronTriggerTimetable as the clock side, the 06:00 tick creates a scheduled run whose logical date is 06:00 and whose data interval starts and ends at 06:00, a point with no span to read. The asset side creates an asset-triggered run with no logical date at all. So the task branches: when triggering_asset_events holds events it reads them, and otherwise it builds the 24 hours ending at the logical date itself.

A producer that writes 1 directory per day, file:///lake/orders/2026-11-10/, cannot declare that path as an outlet: the path differs in every run, and the DAG file is parsed before any run exists. An AssetAlias is a fixed name a task can attach any asset to while it runs. The task declares the alias in outlets, and inside its body adds the concrete asset to outlet_events[alias]. When the task succeeds, Airflow records an event on that concrete asset, records that the alias produced it, and triggers 2 kinds of consumer: DAGs scheduled on the alias, which fire for any asset added through it, and DAGs scheduled on that exact path.

After the run for Nov 10 succeeds, the Assets page shows file:///lake/orders/2026-11-10/ as a real asset with daily_orders as its producer. any_partition_report runs for it as it runs for every day, and one_partition_report runs for this day only. Airflow records the alias-to-path resolution at the event, and the dependency lines in the UI catch up at the next parse of the file. A task that adds 4 assets through one alias writes 4 events, 1 per concrete asset, all at its success, so a consumer scheduled on the alias gets 1 run whose triggering_asset_events lists all 4, each event naming its own concrete path.

A close-of-day run for a trading desk happens at 18:00 New York on each day the market is open, and its window runs from the previous close to this one. Weekdays are a cron matter. The 9 or so exchange holidays a year are a list no cron string holds, and the window after a Thursday holiday has to span 48 hours. When neither built-in cron timetable can say when the next window ends, you write the rule as a Timetable subclass, and the scheduler calls it exactly the way it calls the built-in ones.

restriction carries the DAG's bounds: earliest is the start_date, latest is the end_date or None, and catchup is the DAG's catchup flag. With catchup on, the first window starts at the first close on or after start_date; with it off, the first window starts at the most recent close before now, the same jump to the latest window the built-in timetables make.

The 2 private helpers walk the calendar a day at a time, which is slow in theory and instant in practice: a loop that skips a 3-day weekend and a holiday runs 5 iterations. serialize returns an empty dict because this timetable has no settings; a version that took the holiday list as a constructor argument would return it there and rebuild from it in deserialize.

The plugin must be present wherever the DAG is read: the DAG processor parses the file and needs the class to build the DAG, the scheduler needs it to call next_dagrun_info, and the API server needs it to show the description in the UI. The serialized DAG stores the class's import path and the dict from serialize, and a timetable class that is not registered through a plugin fails at parse time, with the DAG shown as broken in the UI.

Thursday Nov 26 is in HOLIDAYS, so _close_after(Wed 18:00) steps past Thursday and returns Friday's close, and run 3 covers 48 hours of trades. The weekend gives run 4 a 72-hour window. The logical date is the window's start, as it is for every data interval timetable, so run 3's logical date is Wednesday evening even though it is created on Friday. A person who triggers the DAG by hand on Saturday gets infer_manual_data_interval(Saturday): the window from Friday's close back to the close before it, the same window run 3 read.

depends_on_past=True is a task setting, set on 1 task or through default_args for all of them. Before the scheduler starts an instance of that task, it looks at the same task in the DAG's previous run, the run with the latest logical date before this one. If that instance reached success or skipped, this one may start. If it failed, or has not finished, this one waits with no state, and the scheduler checks again on every loop. The first run of the DAG has no previous run and starts without waiting, and the tasks without the setting in a later run proceed as usual.

Say the store the rollup writes to is down from 02:00 to 08:00. The rollup created at 02:00 fails; the 6 runs created from 03:00 to 08:00 arrive on time, their extract instances succeed, and their 6 rollups wait. No other task fails, since the tasks after each waiting rollup never start. At 08:10 the 02:00 rollup is cleared and succeeds, and the 6 waiting rollups run back to back, oldest first, each reading the count the previous one wrote. With depends_on_past=False, the 6 rollups that started inside the outage would have failed, the one created at 08:00 would have succeeded on a count missing 6 hours, and clearing the failed 6 would run them in parallel, each reading whatever count happened to be there.

wait_for_downstream=True tightens the rule by 1 step: the task waits until the previous run's instance of itself and of every task immediately downstream of it have reached success or skipped. It covers the case where the previous run's rollup succeeded but its publish did not, and this run's rollup would otherwise overwrite rows publish has not read yet. Setting it turns depends_on_past on for that task too.

A deployed DAG runs hourly, and a request arrives to make it daily. You edit schedule, the DAG processor parses the file on its next pass, and the new serialized DAG replaces the old one as a new DAG version. The runs that already exist stay exactly as they are: their windows, logical dates and states came from the old timetable, and nothing rewrites them. The open question is where the new timetable puts the next run, and the answer comes from the method every timetable implements.

The scheduler hands the new timetable last_automated_data_interval, the window of the latest run it created itself (scheduled or backfill, never manual or asset-triggered), and reads the next window from next_dagrun_info. The built-in data interval timetables align that window's end down to the previous tick of the new schedule and start the next window there. That one rule, plus the fact that 2 runs of a DAG cannot share a logical date, explains every case in the table.

Hourly to daily is the switch that loses data. The aligned next window is Nov 10 00:00 to Nov 11 00:00, whose logical date is Nov 10 00:00, and the run scheduled__2026-11-10T00:00:00+00:00 already exists: the hourly run for the first hour of the day holds that id. When the window closes at midnight, the scheduler finds the logical date taken, creates no run, writes a log line saying so, and moves the DAG's next window on to Nov 11 to Nov 12. Hours 00:00 to 10:00 were processed by hourly runs, and hours 10:00 to midnight by nobody. No dialog produces a 14-hour window for a daily timetable, so the recovery is a run of the task by hand over those hours, and the cheap path is to deploy the switch at a moment where no collision can happen.