Orchestration concepts: DAGs, schedules, backfills

Seven scripts, one crontab and an email to ops at 08:00: that is how many teams run their first data pipelinedata pipeline: A sequence of jobs that moves data from sources to destinations, with transformations along the way. More in “What data engineering is, and what it is not”. The scripts start at fixed minutes and trust that the one before has finished, so on the morning the menu export lands 25 minutes late, the totals come from an empty file, and nobody can say which day to run again, or whether running it twice is safe.

An orchestrator replaces that trust with rules, built here on a canteen chain’s seven-task pipeline. By the end you can order a pipeline’s tasks as a graph, choose what starts each run, work out the slice of time a run must process, set retries, timeouts and alerts with numbers, and decide when one cron line is still enough.

A pipeline is a DAG of tasks, run on a schedule

A canteen chain with kitchens in Pune, Indore and Kochi wants one summary each morning, and seven tasks make it. extract_orders and extract_menu read the day’s two files. validate prices each order from the menu, so it needs both. aggregate_by_city adds up what is left and write_partition stores the totals under the day’s date. Then send_summary emails ops and archive_source moves the files away.

Draw each task as a box and each “needs” as an arrow from the task that must finish first, and you have a directed acyclic graphdirected acyclic graph: A graph of tasks whose edges have a direction and never form a cycle. The orchestrator can order its tasks and detect a dependency loop., the structure Airflow names its pipelines after: a Dag holds the schedule, the tasks and their dependencies, and leaves what happens inside a task to the task (Airflow: Dags). Acyclic means no path of arrows leads from a task back to itself, and that is what makes an order possible: a complete topological order, in which every task comes after all the tasks it depends on, exists exactly when the graph has no directed cycle (Python: graphlib).

Step through one day’s run and watch which tasks are ready. A task becomes ready only when every arrow into it starts at a task that has finished.

The two extracts started together, and so did the last two tasks. That one rule is all a schedulerscheduler: The part of an orchestrator that decides when each run of each DAG starts and queues its tasks. needs. The program below applies it by hand: it counts the upstream tasks each task still waits for, and a task joins ready when its count reaches 0.

# The order in which a scheduler can start the canteen tasks
downstream = {  # each task, and the tasks that wait for it
    "extract_orders": ["validate"],
    "extract_menu": ["validate"],
    "validate": ["aggregate_by_city"],
    "aggregate_by_city": ["write_partition"],
    "write_partition": ["send_summary", "archive_source"],
    "send_summary": [],
    "archive_source": [],
}
waiting = dict.fromkeys(downstream, 0)  # upstream tasks not yet done
for task in downstream:
    for nxt in downstream[task]:
        waiting[nxt] += 1
ready = [t for t in downstream if waiting[t] == 0]
order = []
while ready:
    task = ready.pop(0)
    order.append(task)
    for nxt in downstream[task]:
        waiting[nxt] -= 1
        if waiting[nxt] == 0:
            ready.append(nxt)
print(len(order), "of", len(downstream), "tasks ordered:")
print(" -> ".join(order))

7 of 7 tasks ordered:
extract_orders -> extract_menu -> validate -> aggregate_by_city -> write_partition -> send_summary -> archive_source

The next program is a small engine for the whole pipeline. It starts a task only after every task it waits for has succeeded, and it runs three times: on a normal day, on a day without the menu export, and after an edit that adds one arrow too many.

"""The canteen pipeline as a DAG, run by a tiny scheduler: a task starts only
when every task it waits for has succeeded. Standard library only."""

from graphlib import CycleError, TopologicalSorter

DAY = "2026-10-16"  # the run's data interval: this whole day, in UTC
ORDERS = [  # (order_id, city, item, quantity)
    (101, "Pune", "thali", 2),
    (102, "Pune", "dosa", 1),
    (103, "Indore", "thali", 1),
    (104, "Indore", "poha", 3),
    (105, "Kochi", "dosa", 2),
    (106, "Kochi", "idli", 0),
]
MENU = {"thali": 120, "dosa": 80, "poha": 40, "idli": 50}
FILES = {f"orders_{DAY}.csv": ORDERS, f"menu_{DAY}.csv": MENU}
TASKS = {}


def task(fn):
    TASKS[fn.__name__] = fn
    return fn


def read(name):
    if name not in FILES:
        raise FileNotFoundError(f"{name} not found")
    return FILES[name]


@task
def extract_orders(out):
    rows = read(f"orders_{DAY}.csv")
    return rows, f"{len(rows)} rows"


@task
def extract_menu(out):
    menu = read(f"menu_{DAY}.csv")
    return menu, f"{len(menu)} items"


@task
def validate(out):
    menu, rows = out["extract_menu"], out["extract_orders"]
    kept = [(o, c, q * menu[i]) for o, c, i, q in rows if i in menu and q > 0]
    return kept, f"{len(kept)} rows kept, {len(rows) - len(kept)} rejected"


@task
def aggregate_by_city(out):
    totals = {}
    for _, city, amount in out["validate"]:
        totals[city] = totals.get(city, 0) + amount
    return totals, ", ".join(f"{c} {t}" for c, t in totals.items())


@task
def write_partition(out):
    path, rows = f"orders_daily/date={DAY}", len(out["aggregate_by_city"])
    return path, f"{path}, {rows} rows"


@task
def send_summary(out):
    total = sum(out["aggregate_by_city"].values())
    return None, f"email to ops: {total} from {len(out['validate'])} orders"


@task
def archive_source(out):
    return None, f"{len(FILES)} source files moved to archive/"


AFTER = {  # each task, and the tasks it waits for (the DAG's edges)
    "extract_orders": [],
    "extract_menu": [],
    "validate": ["extract_orders", "extract_menu"],
    "aggregate_by_city": ["validate"],
    "write_partition": ["aggregate_by_city"],
    "send_summary": ["write_partition"],
    "archive_source": ["write_partition"],
}


def run(after):
    sorter = TopologicalSorter(after)
    try:
        sorter.prepare()  # checks the whole graph before any task runs
    except CycleError as err:
        print(f"  refused to start, {err.args[0]}: {' -> '.join(err.args[1])}")
        return
    out, failed, ready = {}, [], list(sorter.get_ready())
    while ready:
        name = ready.pop(0)
        try:
            out[name], note = TASKS[name](out)
        except Exception as err:  # any error fails the task
            failed.append(name)
            print(f"  FAILED   {name:<18} {err}")
            continue  # a failed task is never marked done
        print(f"  ok       {name:<18} {note}")
        sorter.done(name)  # may make the tasks that wait for it ready
        ready += sorter.get_ready()
    for name in after:
        if name not in out and name not in failed:
            print(f"  not run  {name:<18} an upstream task failed")
    print(f"  {len(out)} of {len(after)} tasks ok")


print(f"Run 1: the run for {DAY}")
run(AFTER)
print("Run 2: the same day, but the menu export has not landed")
del FILES[f"menu_{DAY}.csv"]
run(AFTER)
print("Run 3: someone makes validate wait for aggregate_by_city too")
run({**AFTER, "validate": AFTER["validate"] + ["aggregate_by_city"]})

Run 1: the run for 2026-10-16
  ok       extract_orders     6 rows
  ok       extract_menu       4 items
  ok       validate           5 rows kept, 1 rejected
  ok       aggregate_by_city  Pune 320, Indore 240, Kochi 160
  ok       write_partition    orders_daily/date=2026-10-16, 3 rows
  ok       send_summary       email to ops: 720 from 5 orders
  ok       archive_source     2 source files moved to archive/
  7 of 7 tasks ok
Run 2: the same day, but the menu export has not landed
  ok       extract_orders     6 rows
  FAILED   extract_menu       menu_2026-10-16.csv not found
  not run  validate           an upstream task failed
  not run  aggregate_by_city  an upstream task failed
  not run  write_partition    an upstream task failed
  not run  send_summary       an upstream task failed
  not run  archive_source     an upstream task failed
  1 of 7 tasks ok
Run 3: someone makes validate wait for aggregate_by_city too
  refused to start, nodes are in a cycle: validate -> aggregate_by_city -> validate

Run 1 ends with 7 of 7 tasks ok and 720 in the email. In run 2, extract_menu fails and is never marked done, so the 5 tasks below it never start and no wrong email goes out. Airflow gives such tasks the state upstream_failed, and fails the whole pipeline runpipeline run: One execution of a pipeline for one data interval or one event. The orchestrator keeps a record of its state, logs and outputs. when any task with nothing after it ends failed or upstream_failed (Airflow: Tasks; Dag Runs).

A run starts on the clock, on new data or on a message

Something has to start each run: a time on the clock, an update to the data the pipeline reads, or a message from another system.

A clock. A cron expressioncron expression: A schedule written as five fields, minute, hour, day of month, month and day of week, such as 30 2 * * * for 02:30 every day. has five fields, and POSIX fixes their order and their ranges: minutes run from 0 to 59, hours from 0 to 23, days of the month from 1 to 31, months from 1 to 12, and days of the week from 0 to 6, counted from Sunday as 0. Each field is * for every value, a number, a range such as 1-5, or a comma-separated list of those (POSIX crontab, IEEE Std 1003.1-2024). Two schedules, field by field:

Two details catch people out. With both day fields set, a day that matches either one runs the job: the standard’s 0 0 1,15 * 1 fires at midnight on the 1st and the 15th of every month, and also on each Monday. And the step form */15 is not among the forms POSIX defines, so check what your cron accepts.

A calendar. A timetable generalises the clock. Airflow turns every cron string into one, and the timetable also decides each run’s data interval (Airflow: Timetables). It can carry a time zone, as CronTriggerTimetable("0 10 * * 1-5", timezone="Asia/Kolkata") does for 10:00 India time on weekdays. Airflow keeps times in UTC and lets such a schedule follow its zone’s clock changes (Airflow: Time Zones).

An update to the data. A data-aware trigger starts a run when its input changes. In Airflow a producing task declares outlets=[Asset("s3://canteen/orders/")] and the consuming Dag uses that asset as its schedule. The asset counts as updated only when the producing task succeeds, and a Dag that names several assets runs once all of them have been updated since its last run (Airflow: Asset-Aware Scheduling).

A message. An event-driven triggerevent-driven trigger: A rule that starts a run when a message or webhook arrives, instead of on a clock. Useful when the data's arrival is irregular. starts a run when an outside system sends word, such as a queue message or a webhook. Airflow 3.0 added an AssetWatcher that watches a source such as a message queue and updates an asset when an event arrives (Airflow: Event-driven scheduling). A queue with at-least-once deliveryat-least-once delivery: A delivery guarantee under which no message is lost but some can arrive more than once, because a sender that gets no acknowledgement sends again; the receiver has to recognise and drop the repeats. may hand over a message twice, so such a run must be safe to start twice.

Answer the two questions in the table about your own input. Each answer rules options out or adds a caveat, with the reason:

In short: a clock for input that is reliably there in time, a data-aware trigger for another pipeline’s output, and an event for input that arrives from outside on a timetable of its own.

Every run processes a defined data interval, not “now”

The run that fires at 00:00 UTC as 2026-10-17 begins must total the orders of 2026-10-16, but at that moment the clock proves nothing: a run can start late, be retried at noon or be started by hand next week. So each run gets a data intervaldata interval: The slice of time a run processes, such as one day from midnight to the next midnight with the end excluded. Reading it instead of the clock keeps re-runs and backfills correct., the slice of time it processes. Airflow’s documentation describes a daily interval as running from one midnight to the next, with the run scheduled after it ends, so the day’s data is complete (Airflow: Dag Runs). The interval selects orders by event timeevent time: The time an event happened at its source, stamped by the system that produced it, such as a payment's capture time; totals by event time put each row on the day it happened., when they were placed.

The program works out one scheduled run’s interval. Then it replays three runs on one afternoon, first reading each run’s interval and then now() minus one day, as a task that ignores its interval would:

"""Which day of orders does a run process? Its data interval, not the clock."""

from datetime import UTC, datetime, timedelta

ONE_DAY = timedelta(days=1)


def daily_interval(fired_at):
    """A daily schedule at 00:00 UTC: a run covers the whole day before it fires."""
    end = fired_at.replace(hour=0, minute=0)
    return end - ONE_DAY, end


def hhmm(t):
    return t.strftime("%Y-%m-%d %H:%M")


fired = datetime(2026, 10, 17, 0, 0, tzinfo=UTC)
start, end = daily_interval(fired)
print(f"The scheduled run fired at {hhmm(fired)} UTC reads")
print(f"  [{hhmm(start)}, {hhmm(end)}) UTC, end excluded")
print(f"  day by its interval:    {start:%Y-%m-%d}")
print(f"  day by now() - 1 day:   {fired - ONE_DAY:%Y-%m-%d} (the same)")

now = datetime(2026, 10, 17, 15, 40, tzinfo=UTC)  # someone starts a replay
print(f"\nA replay of 3 runs, all started by hand at {hhmm(now)} UTC")
print(f"{'run':<5}{'data interval it was given':<38}{'by interval':<13}now() - 1 day")
days = {"by interval": [], "by now() - 1 day": []}
for n, d in enumerate((15, 16, 17), start=1):
    start, end = daily_interval(datetime(2026, 10, d, tzinfo=UTC))
    by_interval, by_clock = f"{start:%Y-%m-%d}", f"{now - ONE_DAY:%Y-%m-%d}"
    days["by interval"].append(by_interval)
    days["by now() - 1 day"].append(by_clock)
    print(f"{n:<5}[{hhmm(start)}, {hhmm(end)})  {by_interval}   {by_clock}")
wanted = set(days["by interval"])
for rule, got in days.items():
    wrote, missed = len(set(got)), len(wanted - set(got))
    unit = "day" if wrote == 1 else "days"
    print(f"{rule:<17} 3 runs wrote {wrote} {unit}, missed {missed} of 3")

The scheduled run fired at 2026-10-17 00:00 UTC reads
  [2026-10-16 00:00, 2026-10-17 00:00) UTC, end excluded
  day by its interval:    2026-10-16
  day by now() - 1 day:   2026-10-16 (the same)

A replay of 3 runs, all started by hand at 2026-10-17 15:40 UTC
run  data interval it was given            by interval  now() - 1 day
1    [2026-10-14 00:00, 2026-10-15 00:00)  2026-10-14   2026-10-16
2    [2026-10-15 00:00, 2026-10-16 00:00)  2026-10-15   2026-10-16
3    [2026-10-16 00:00, 2026-10-17 00:00)  2026-10-16   2026-10-16
by interval       3 runs wrote 3 days, missed 0 of 3
by now() - 1 day  3 runs wrote 1 day, missed 2 of 3

The scheduled run reads [2026-10-16 00:00, 2026-10-17 00:00), end excluded, and on its own night now() minus a day picks the same day, which is why the bug hides. The replay exposes it: by interval, 3 runs wrote 3 days; by the clock, they wrote 1 day and missed 2 of 3. A backfillbackfill: Running a pipeline over past intervals to fill a gap or to reprocess data after a bug fix. A backfill is only safe when every task is idempotent. is that replay done on purpose: Airflow can run a copy of a Dag for each day of the past three months at once, each looking only at its own interval (Airflow: Dags). Its best-practice guide asks tasks to read and write one partition, such as one keyed by data_interval_start, and never to use now() for critical work (Airflow: Best Practices).

Intervals also tell the scheduler what is missing after a late start or a pause. With catchupcatchup: The scheduler's behaviour of creating a run for every interval missed since the start date or a pause. With it off, the scheduler creates a run only for the latest interval. on, it creates a run for every interval since the start date that has not run yet; Airflow’s default is off, which creates a run only for the latest interval (Airflow: Dag Runs).

Retries absorb brief failures, and timeouts cap how long a task can block

extract_orders calls the canteen’s order service, which now and then answers HTTP 503 for a minute or hangs on a slow query. A retry absorbs the first kind of failure and a timeout ends the second. Airflow sets both per task (Airflow Task SDK: API reference; Airflow: Tasks):

Doubling the wait is exponential backoffexponential backoff: Waiting longer before each retry of a failed request, usually by doubling a ceiling from a base wait up to a cap, so that a struggling service gets time to recover., which gives a struggling service time to recover; jitterjitter: Randomness in retry waits, such as a wait drawn between zero and the backoff ceiling, so that clients that failed together do not all retry at the same moment. keeps many clients from retrying in step. The program works out how long these settings can hold up validate, with every attempt running into its timeout, then reads a week of runs:

"""How long can extract_orders hold up validate, and which alert sees a retry?"""

TIMEOUT_S = 5 * 60  # execution_timeout: each attempt is stopped after 5 minutes
RETRIES = 3  # so up to 4 attempts in all
BASE_DELAY_S = 30  # retry_delay before the first retry
MAX_DELAY_S = 5 * 60  # max_retry_delay: no wait is longer than this


def worst_case(waits):
    """Every attempt runs into the timeout, and each retry waits first."""
    return (len(waits) + 1) * TIMEOUT_S + sum(waits)


doubled = [min(BASE_DELAY_S * 2**k, MAX_DELAY_S) for k in range(RETRIES)]
plans = {
    "fixed 30 s": [BASE_DELAY_S] * RETRIES,
    "doubling": doubled,
    # Airflow 3.3.2 spreads a doubled wait d over d to 2d - 1 seconds
    "doubling, spread": [min(2 * d - 1, MAX_DELAY_S) for d in doubled],
}
head = f"{RETRIES + 1} attempts of up to {TIMEOUT_S // 60} min"
print(f"Worst case before validate can start ({head})")
for name, waits in plans.items():
    total = worst_case(waits)
    print(f"  {name:<16} waits {waits} s -> {total // 60} min {total % 60:02d} s")

WEEK = {  # what each attempt of extract_orders did, one run a day
    "Mon": ["ok"],
    "Tue": ["timeout", "ok"],
    "Wed": ["ok"],
    "Thu": ["HTTP 503", "HTTP 503", "ok"],
    "Fri": ["ok"],
    "Sat": ["ok"],
    "Sun": ["timeout", "ok"],
}
print("\nOne week of runs")
for day, attempts in WEEK.items():
    print(f"  {day}  {' -> '.join(attempts)}")
failed = sum(attempts[-1] != "ok" for attempts in WEEK.values())
retries = sum(len(attempts) - 1 for attempts in WEEK.values())
days = sum(len(attempts) > 1 for attempts in WEEK.values())
print(f"{len(WEEK) - failed} of {len(WEEK)} runs succeeded")
print(f"  alert on failure only: {failed} alerts")
print(f"  alert on every retry:  {retries} alerts, on {days} of {len(WEEK)} days")

Worst case before validate can start (4 attempts of up to 5 min)
  fixed 30 s       waits [30, 30, 30] s -> 21 min 30 s
  doubling         waits [30, 60, 120] s -> 23 min 30 s
  doubling, spread waits [59, 119, 239] s -> 26 min 57 s

One week of runs
  Mon  ok
  Tue  timeout -> ok
  Wed  ok
  Thu  HTTP 503 -> HTTP 503 -> ok
  Fri  ok
  Sat  ok
  Sun  timeout -> ok
7 of 7 runs succeeded
  alert on failure only: 0 alerts
  alert on every retry:  4 alerts, on 3 of 7 days

With a fixed 30 s wait the worst case is 4 attempts of 5 minutes plus 3 waits, 21 min 30 s, and doubling the wait gives 23 min 30 s. Airflow 3.3.2 adds jitter of its own: its code spreads each doubled wait over a range that starts at that value and ends one second short of twice it, from a hash of the task, its run and the try, so the spread line’s 26 min 57 s is the real bound (Airflow 3.3.2 source). All three fit well inside the hours before the 08:00 email. Set RETRIES to 5 and run it again: the bound grows, and the 5-minute cap stops the last waits from growing further.

The week shows why failure alerts are not enough. All 7 runs succeeded, so none fired, yet the task needed 4 retries on 3 days. A source that needs retries that often is degrading, and on the day it needs one attempt more than the budget, the run fails without warning. Alert on every retry, and alert on lateness too: a run can miss the 08:00 email with no failed task at all, just by being slow. That 08:00 is a service-level objectiveservice-level objective: A target for a measured property of a service, with how often it must hold, such as a file published by 06:00 on at least 29 of 30 days. A promise to customers with a consequence for missing it is a service level agreement. More in “Pipeline requirements: SLAs, freshness, sizing” on data freshnessdata freshness: How old data is when someone reads it, or how soon after an event it can be read. Written as a target, for example that the oldest record is no older than a stated time. More in “Pipeline requirements: SLAs, freshness, sizing”, and once the canteen promises it to someone outside the team, with a consequence when it is missed, it is a service level agreement (SLA). Airflow 3.0 removed the SLA feature of Airflow 2, and 3.1 added Deadline Alerts, still experimental, which run a callback when a run has not finished within a set time of a reference point such as its queueing (Airflow: Tasks; Deadline Alerts). A deadline turns either kind of promise into an alert.

When cron is enough, and when it is not

Some pipelines never need more than 30 2 * * * /opt/canteen/daily.sh: one script on one machine writing one report for one reader. Give it a log, make it exit with an error status when anything fails, and let errors reach a person: cron mails a job’s output and errors to its owner unless the job redirects them (POSIX crontab). That is a schedule, a log and an alert; an orchestrator would add a server, a database and upgrades to keep three lines of shell running.

Cron stops being enough when jobs need each other. It knows start times only, so a dependency becomes a guess (“the menu job is always done by 02:30”), and it records nothing about which day a run processed, so a backfill is a loop someone writes by hand. Each row below is a point where an orchestrator starts to pay:

The orchestrator’s job is the order, not the work: Airflow’s documentation says a Dag cares how its tasks run, in what order, with how many retries and which timeouts, not what they do (Airflow: Dags). Heavy processing belongs to the engines tasks call, such as a warehouse or Spark. Orchestrators differ in what they add on top: Dagster, for one, calls itself a data orchestrator with integrated lineage and observability (Dagster documentation).

Next, Writing a TaskFlow DAG and testing it with airflow dags test turns the canteen pipeline into an Airflow 3 Dag; Idempotent tasks and backfills makes its writes safe to repeat and backfills a date range with airflow backfill create; and Dagster, Prefect, Kestra and cron compared runs the same pipeline in other tools.

Practise: write the canteen DAG as YAML

Every orchestrator file starts by writing the graph down. In the exercise you write the seven canteen tasks as YAML, each with the tasks it waits for and its retry rules, in an order a scheduler could follow from the top. The sample tests check the tasks, the order, the six edges and the extracts’ retries.

"""Write the canteen pipeline as a YAML DAG that a scheduler can run in order.

Fill in DAG_YAML below and leave parse_dag as it is: the tests read your YAML
with parse_dag, then check the order, the dependencies and the retry rules.
"""

# Each task is a key at the left edge. Under it, indented by two spaces:
#   after          the tasks it waits for, as a list: [task_a, task_b]
#   retries        how many times a failed attempt is tried again
#   retry_delay_s  seconds to wait before each retry
#   timeout_s      seconds after which an attempt is stopped
# The tasks are listed here in A-to-Z order, with nothing filled in yet.
DAG_YAML = """
aggregate_by_city:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
archive_source:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
extract_menu:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
extract_orders:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
send_summary:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
validate:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
write_partition:
  after: []
  retries: 0
  retry_delay_s: 0
  timeout_s: 0
"""


def parse_dag(text):
    """Reads the YAML above into {task: {"after": [...], "retries": 3, ...}},
    keeping the tasks in the order they are written."""
    dag, task = {}, None
    for raw in text.splitlines():
        line = raw.split("#")[0].rstrip()
        if not line:
            continue
        if not line.startswith(" "):
            task = line.rstrip(":").strip()
            dag[task] = {}
            continue
        if task is None:
            raise ValueError(f"{line.strip()!r} is indented but has no task above it")
        key, _, value = (part.strip() for part in line.partition(":"))
        if value.startswith("[") and value.endswith("]"):
            dag[task][key] = [x.strip() for x in value[1:-1].split(",") if x.strip()]
        elif value.isdigit():
            dag[task][key] = int(value)
        else:
            raise ValueError(f"{task}: {key} must be a [list] or a whole number")
    return dag

from canteen_dag import DAG_YAML, parse_dag

TASKS = {
    "extract_orders",
    "extract_menu",
    "validate",
    "aggregate_by_city",
    "write_partition",
    "send_summary",
    "archive_source",
}
NEEDED = {
    ("extract_orders", "validate"),
    ("extract_menu", "validate"),
    ("validate", "aggregate_by_city"),
    ("aggregate_by_city", "write_partition"),
    ("write_partition", "send_summary"),
    ("write_partition", "archive_source"),
}


def edges(dag):
    """Every (upstream, task) pair the YAML declares."""
    return {(up, task) for task, spec in dag.items() for up in spec.get("after", [])}


def test_the_seven_tasks_are_all_listed():
    """the YAML holds the seven canteen tasks and no others."""
    assert set(parse_dag(DAG_YAML)) == TASKS


def test_every_task_is_listed_after_the_tasks_it_waits_for():
    """reading the YAML from the top gives a topological order."""
    seen = []
    for task, spec in parse_dag(DAG_YAML).items():
        for up in spec.get("after", []):
            assert up in seen, f"{task} is listed before {up}, which it waits for"
        seen.append(task)


def test_no_dependency_is_missing():
    """the six arrows of the lesson's pipeline are all there."""
    missing = NEEDED - edges(parse_dag(DAG_YAML))
    assert not missing, f"missing (upstream, task) pairs: {sorted(missing)}"


def test_the_extracts_retry_3_times_30_s_apart_with_a_5_minute_timeout():
    """both extract tasks: retries 3, retry_delay_s 30, timeout_s 300."""
    dag = parse_dag(DAG_YAML)
    for task in ("extract_orders", "extract_menu"):
        spec = dag[task]
        got = (spec.get("retries"), spec.get("retry_delay_s"), spec.get("timeout_s"))
        assert got == (3, 30, 300), f"{task} has {got}"

Then practise all five ideas; each round starts with the one your answers show you know least.

Interview questions

Two warm-ups first, answered in full. Say your own answer aloud before you read each one, then compare.

Warm-up: how would you explain a DAG to a developer who has only used cron?

Cron answers one question per line: when does this command start? A DAG answers a second one: what has to finish before it starts? Draw each step as a box and each “needs” as an arrow, and the arrows give an order. In the canteen pipeline, validate needs both extracts, so it starts when both are done, not at 02:30 because they usually are by then. Acyclic means no chain of arrows returns to where it began, so an order always exists, and a tool can check the graph before the first run. The scheduler starts each task as soon as its upstream tasks succeed, runs independent tasks side by side, and stops everything downstream of a failure. With cron, a late file means the next job runs on bad input; with a DAG, that job waits, or the run fails where everyone can see it.

Warm-up: what is a data interval, and why does an orchestrator need it?

A data interval is the slice of time one run is responsible for, such as a day from midnight to the next midnight with the end excluded. The scheduler hands it to each run, and every task reads and writes that slice instead of asking the clock. It is needed because runs rarely happen when their data happened: a run starts after its day has ended, a retry can start hours later, and a backfill starts today for days long past. A task that reads the clock processes the wrong day in each of those cases. A task that reads its interval processes the same day every time, so a re-run replaces the same partition. Intervals also show the scheduler what is missing: with catchup on, Airflow creates a run for every interval since the start date that has not run yet.

The next two need a full interview answer. Each shows what it tests and a first step; plan yours around the canteen pipeline before you open the model answer.

Key takeaways

• A pipeline is a directed acyclic graph of tasks: a task starts only when every task it waits for has succeeded, so a missing menu file stops the five tasks below it and a cycle is refused before anything runs.
• A run starts on the clock (a five-field cron expression, or a timetable in a time zone), on an update to its input, or on an outside message, and the way the input arrives decides which fits.
• Each run processes its own data interval; in the replay, runs that read the clock instead wrote 1 day and missed 2 of 3, which is why reruns and backfills depend on intervals.
• Retries absorb brief failures and timeouts cap how long a task blocks: 3 retries 30 s apart with a 5-minute timeout hold up the next task 21 min 30 s at most; an alert on every retry sees a source getting worse, and a deadline on the run reports a slow run that threatens the 08:00 target of an SLO or an SLA.
• Cron and a script are enough for one job on one machine with a log and an alert; an orchestrator pays once jobs depend on each other, past days must run again, or several people share the on-call.