Chapter 24: Apache Airflow: Scheduling, Dependencies, and Managing Complex Pipeline DAGs

"Cron runs things at times. An orchestrator runs things after other things. The difference is everything you learn the hard way."

Overview

Every chapter so far has produced something that has to run: an extract, a CDC consumer, a dbt build, a Spark job, a reconciliation, a quarantine replay. This chapter is about the thing that decides when.

Airflow is the default answer and has been for most of a decade. It is not the best-designed tool in this book and it is the one you are most likely to meet, which is a common enough combination in this field to be worth naming rather than apologizing for.

The chapter is organized around a claim: most Airflow pain is not about Airflow. It is about three things that would be hard in any orchestrator — what a schedule means, what happens on a retry, and what shares capacity with what — plus one thing that is genuinely Airflow's fault, which is §24.3's date model.

Chapter 19 §19.1 said dbt is not an orchestrator and pointed here. Chapter 20 §20.4 asked for a lock so a backfill cannot race the nightly run; that is §24.8. Chapter 23 §23.9 asked for a quarantine replay that is schedulable and idempotent; that is §24.5.

24.1 When Cron Stops Working

Cron is excellent. It is reliable, it is on every machine, it has one job, and for a single independent task it is the right answer — this book has recommended it more than once.

It stops working at the fourth dependency, and the failure is gradual enough that most teams pass the point without noticing.

# The shape every data team writes before it has an orchestrator.
0 2 * * *   /opt/kestrel/extract_orders.sh
15 2 * * *  /opt/kestrel/extract_customers.sh
30 2 * * *  /opt/kestrel/load_bronze.sh
0 3 * * *   /opt/kestrel/dbt_build.sh
0 4 * * *   /opt/kestrel/export_reports.sh

Every one of those times is a guess about how long the previous step takes, and the guesses encode a dependency graph in a form where it cannot be read, tested, or changed safely.

Five things this cannot do, in the order teams hit them:

Run a step because the previous one finished, rather than because fifteen minutes elapsed. On the night the extract takes seventeen minutes, load_bronze reads a partial file.

Retry a failed step without re-running the ones that succeeded. Chapter 20's backfills, and every morning where one task in nine failed.

Say what failed and what was skipped because of it. Cron mails you exit 1, five times, from five jobs, and the causal order is yours to reconstruct.

Fan out over a variable list. Thirteen sources today, fourteen tomorrow.

Tell you what ran, when, and how long it took. Chapter 21 Case Study 2's duration regression is invisible without a history, and cron keeps none.

📐 Design Decision — the honest threshold, and the honest cost

Adopt an orchestrator when you have more than about three dependent steps, or when any step's duration varies enough that a fixed offset is a guess. Below that, cron plus a well-written script is less machinery and fewer things to operate.

And be honest about what an orchestrator costs, because the writing on this topic is not:

• A scheduler, a metadata database, a web server, and workers to run and monitor. Managed options (MWAA, Composer, Astronomer) move the operation rather than removing it, and cost real money — Kestrel's managed Airflow is about $310 a month before any task runs.
• A new failure surface. The orchestrator can be down while every pipeline is healthy, and the symptom is identical to every pipeline being broken.
• A place for logic to accumulate where it does not belong. §24.10.

The version of this argument that is wrong is "you need Airflow because you are a data team." The version that is right is "you have nine tasks with a dependency graph, and it currently lives in five cron offsets and one person's head."

24.2 The Model: DAG, Task, Operator, Executor

Four nouns, and the confusion between the last two costs people a day.

A DAG is a Python file describing tasks and their dependencies. It is not the running thing — it is the definition, parsed repeatedly by the scheduler.

A task is one node. A task instance is one node on one run, and that is the thing that succeeds, fails, and gets retried.

An operator is what a task does. PythonOperator, BashOperator, KubernetesPodOperator, plus a large ecosystem of provider packages.

The executor decides where tasks run. LocalExecutor on the scheduler machine; CeleryExecutor on a worker pool; KubernetesExecutor as one pod per task.

from airflow.decorators import dag, task
import pendulum

@dag(
    dag_id="kestrel_daily",
    schedule="0 3 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                    # §24.4 -- read that section before changing this
    max_active_runs=1,                # §24.8 -- and this one
    default_args={"retries": 2, "retry_delay": pendulum.duration(minutes=5)},
)
def kestrel_daily():

    @task
    def extract_orders(data_interval_start=None, data_interval_end=None):
        ...

    @task
    def load_bronze(paths: list[str]):
        ...

    load_bronze(extract_orders())      # the dependency IS the function call

kestrel_daily()

The TaskFlow API — the @task decorator — is the modern way to write this. The dependency graph comes from the call structure rather than from >> operators, which is the same idea as dbt's ref(): the graph is derived from the code rather than maintained alongside it.

⚠️ Failure Mode — the DAG file is executed on every parse

Airflow parses every DAG file every thirty seconds by default, in the scheduler process. Whatever is at module level runs each time.

```python

At module scope. This runs every 30 seconds, forever, on the scheduler.

df = pd.read_parquet("s3://lake/gold/dim_customer/") # 1.2 GB CUSTOMERS = df["customer_id"].tolist()

@dag(...) def bad(): for c in CUSTOMERS: # and it decides the graph ... ```

Three consequences, and the third is the one that produces a support ticket:

• The scheduler does your I/O, thousands of times a day, and its CPU and memory become a function of your DAG files rather than of your pipelines.
• Parse time delays scheduling for every DAG, not just this one. A slow file is a global problem.
• The graph changes without a deploy, silently, whenever the query returns something different — so a task disappears from history and nobody changed any code.

The rule: a DAG file should import, define, and nothing else. Anything that touches a network, a database, or a large file belongs inside a task.

The check: airflow dags list-import-errors, and then time the parse: time python your_dag.py. Under a second. If it is not, something at module level is doing work.

24.3 The Date Model, Which Is Airflow's Worst Feature

Airflow schedules intervals, not points in time, and the consequence surprises everyone exactly once.

A DAG with schedule="@daily" and a data interval of 2026-03-17 runs at the END of that interval — just after midnight on 2026-03-18.

 data_interval_start        data_interval_end        actually runs
 2026-03-17 00:00           2026-03-18 00:00         2026-03-18 00:00
 └──────── the data this run is about ────────┘      └── when ──┘

The reasoning is sound: a run that processes a day's data cannot start until the day is over. The confusion is that the run is named after the interval's start, so a run labelled 2026-03-17 executes on the 18th and produces the 17th's numbers.

What to use, and what to avoid:

@task
def extract(data_interval_start=None, data_interval_end=None):
    # ✅ The window this run is responsible for. Stable across retries and
    #    reruns -- which is what makes the task idempotent (§24.5).
    rows = query(f"WHERE updated_at >= '{data_interval_start}' "
                 f"  AND updated_at <  '{data_interval_end}'")

    # ❌ Never. This is "whenever this happened to run", so a retry three
    #    hours later processes a different window and a rerun next month
    #    processes next month.
    today = datetime.now().date()

⚠️ Failure Mode — datetime.now() in a task destroys reruns

This is the single most consequential line of code in this chapter.

A task that derives its window from the wall clock is not a function of its inputs. It produces different output every time it runs, which means:

• A retry processes a different window than the attempt it is retrying — so a failure at 03:00 retried at 03:20 may skip twenty minutes of data or reprocess it, depending on the query.
• A rerun cannot reproduce history. Clearing a failed task from last March reprocesses today, writing today's data into March's partition. This is worse than the original failure and it looks like a successful recovery.
• The idempotency work in Chapter 20 §20.3 is defeated. A merge is idempotent on its key; if the window moves, the key set moves, and re-running writes a different set of rows.

data_interval_start and data_interval_end are stable across every retry and every rerun, forever, because they are properties of the run rather than of the moment. That is the whole point of the date model, and it is why the confusing design is the right design.

Grep your DAGs for now(), today(), and utcnow(). Every hit is either a bug or needs a comment explaining why it is not.

🧭 Airflow 3 renames some of this — see §24.13's version note — but the semantics are unchanged and the rule above survives verbatim.

24.4 Catchup and Backfill

catchup=False    # only the most recent interval. The right default.
catchup=True     # every interval since start_date, in order.

catchup=True with a start_date two years ago will schedule 730 runs the moment you deploy it. This is documented, it is what the name says, and it happens to somebody on every team.

When catchup=True is right: the DAG processes discrete, independent time windows and you genuinely want the history filled. When it is wrong: the DAG maintains a current state — a full refresh, a snapshot, a sync — where running it 730 times means running it 730 times for nothing.

A deliberate backfill is a command, not a configuration:

airflow dags backfill kestrel_daily \
    --start-date 2026-04-01 --end-date 2026-04-28

And it is subject to §24.8's concurrency, which is the difference between a backfill and an outage.

24.5 Idempotency, Retries, and Clearing

Chapter 20 §20.3's four strategies apply unchanged. The orchestrator adds one requirement: a task will be re-run, by a retry or by a human clearing it, and you do not control when.

@task(retries=3, retry_delay=pendulum.duration(minutes=5),
      retry_exponential_backoff=True)
def load_bronze(data_interval_start=None, data_interval_end=None):
    # Delete-then-insert on the interval. Chapter 20's Strategy 1, and the
    # reason it works here is that the interval is stable across retries.
    conn.execute("DELETE FROM bronze.orders WHERE dt = %s", (data_interval_start.date(),))
    conn.execute("INSERT INTO bronze.orders SELECT ... WHERE ...", ...)

🔁 Idempotency Check — retries make non-idempotency a certainty rather than a risk

Without an orchestrator, a non-idempotent job is a hazard: it will double-count if something re-runs it.

With retries=3, it will double-count. Not might. A transient network error at 03:00 is a weekly event, and the retry that follows is the thing your appending task was waiting for.

Three specific hazards the retry mechanism introduces:

A task that succeeded but reported failure. A write completes and the connection drops before the acknowledgment. The retry runs against a target that already has the data. This is not rare — it is the normal failure mode of a network — and only idempotency covers it.

A task that partially completed. Wrote three of five partitions, then failed. The retry must handle a target in a state no clean run produces.

A human clearing a task from a month ago. §24.3's rule is what makes this safe, and its absence is what makes it destructive.

The test, and it is one command:

bash airflow tasks test kestrel_daily load_bronze 2026-03-17 airflow tasks test kestrel_daily load_bronze 2026-03-17 # again

Then diff the target. airflow tasks test runs a task outside a DAG run, against a real interval, without touching the metadata database — which makes it the right tool for exactly this and is used far less than it should be.

Clearing is Airflow's recovery primitive: mark a task instance (and optionally its downstream) as not-yet-run, and the scheduler re-runs it. It is the reason DAG structure matters, which is §24.6.

24.6 Structure: Recovering From the Middle

A DAG's structure determines what a 04:00 recovery costs.

ONE BIG TASK                    SPLIT BY STEP
┌──────────────────┐            ┌────────┐  ┌──────┐  ┌───────┐  ┌────────┐
│ extract, load,   │            │extract │→ │ load │→ │ dbt   │→ │ export │
│ transform, export│            └────────┘  └──────┘  └───────┘  └────────┘
└──────────────────┘
a failure in `export`           a failure in `export` clears
re-runs the extract              and re-runs `export`
(47 min)                         (90 s)

But splitting has a cost, and the usual advice omits it: every task boundary is a scheduling round trip, worth a few seconds of latency and a row in the metadata database. A DAG with 4,000 one-second tasks spends more time being scheduled than working, and its scheduler load is real.

The judgment: a task should be the smallest unit you would want to retry independently. Which is Chapter 19 §19.5's model-size rule, applied to a different graph — and both come out at "the unit of failure is the unit of design."

Task groups organize without adding boundaries:

with TaskGroup("extract") as extract:
    for source in SOURCES:            # a LITERAL list. §24.2.
        extract_one.override(task_id=f"extract_{source}")(source)

Dynamic task mapping fans out over a value computed at run time, which is the supported way to do what §24.2 forbids at module level:

@task
def list_files(data_interval_start=None) -> list[str]:
    return s3.list(f"raw/dt={data_interval_start.date()}/")

@task
def process(path: str):
    ...

process.expand(path=list_files())     # one task instance per file

24.7 Sensors, Deferrable Operators, and Datasets

Waiting for something is a large fraction of what orchestration does, and Airflow has three answers of increasing quality.

A sensor polls. S3KeySensor, ExternalTaskSensor, SqlSensor. Simple, and it occupies a worker slot the entire time it waits.

A deferrable operator releases its slot. It registers interest with a triggerer process and suspends; the triggerer watches thousands of conditions in one async event loop and wakes the task when one fires.

S3KeySensor(task_id="wait_for_extract", bucket_key="...",
            deferrable=True)          # ← the whole change

💸 Cost Check — sixteen worker slots spent waiting

Kestrel had sixteen ExternalTaskSensor tasks waiting on upstream DAGs, each polling every sixty seconds, each holding a worker slot for an average of 41 minutes.

$$16\ \text{sensors} \times 41\ \text{min} = 656\ \text{slot-minutes a night}$$

The worker pool has 32 slots. For roughly the first hour of the nightly window, half the pool was occupied by tasks doing nothing, and the real jobs queued behind them.

The visible symptom was not cost. It was that the DAG finished at 04:50 instead of 03:55, and the investigation went to the jobs because the sensors showed as "running."

deferrable=True on all sixteen returned 656 slot-minutes a night, and the DAG finished at 03:58. One keyword, on sixteen lines.

And the deeper fix, which they also did: twelve of the sixteen sensors existed to express a dependency between DAGs, which is what datasets are for — see below. A sensor that waits for another DAG is a dependency written as a poll.

The general shape: a task that is "running" but doing nothing is invisible to every dashboard, because every dashboard measures failure and duration, and a sensor is neither failing nor slow.

⚠️ Failure Mode — ExternalTaskSensor waits for the wrong run

An ExternalTaskSensor looks for a task instance in the other DAG at the same logical date, and if the two DAGs are on different schedules there is no such instance. The sensor then waits until its timeout, succeeds nothing, and fails — reporting an upstream failure that did not happen.

python ExternalTaskSensor( task_id="wait_for_hourly", external_dag_id="kestrel_hourly", # WITHOUT this, a daily DAG looks for an hourly run at 00:00 only. execution_delta=timedelta(hours=1), # or execution_date_fn=... deferrable=True, )

The offset is a second place the schedule relationship is written down, and it drifts the moment either schedule changes — which is the strongest single argument for the datasets below. A dependency expressed as an arithmetic offset between two cron expressions is a dependency waiting to be wrong.

Datasets (Airflow 2.4+) invert the relationship. Instead of DAG B waiting for DAG A, DAG A declares what it produces and DAG B declares what it consumes, and Airflow schedules B when A updates it.

ORDERS = Dataset("s3://lake/silver/orders/")

@dag(schedule="0 3 * * *")             # producer: time-triggered
def extract_orders():
    @task(outlets=[ORDERS])
    def load(): ...

@dag(schedule=[ORDERS])                # consumer: data-triggered
def build_marts():
    ...

This is the same idea as ref() and TaskFlow's call structure, a third time: the dependency is declared once, on the thing itself, and the schedule is derived. It removes an entire class of "DAG B ran before DAG A finished" bug, and it is the right default for cross-DAG dependencies in any Airflow new enough to have it.

24.8 Pools and Concurrency: The Lock Chapter 20 Asked For

Four separate limits, and knowing which one is biting is most of the debugging.

Pools are the mechanism Chapter 20 §20.4 wanted. A backfill and a nightly run that both write fct_order_item must not overlap, and a pool of one enforces it:

@task(pool="warehouse_writes", pool_slots=1)
def build_fct_order_item(...):
    ...

max_active_runs=1 is the other half, and it is the more commonly forgotten one: without it, a run that overruns is joined by the next scheduled run, and both write the same target.

⚠️ Failure Mode — the backfill that starved the nightly run

A 90-day backfill was started at 16:00. Each run took about eleven minutes and Airflow ran as many concurrently as the pool allowed — which was the default pool, with 128 slots.

At 03:00 the nightly DAG's tasks queued behind 61 backfill tasks. The nightly run started at 05:12 and finished at 06:41. The 6am SLA was missed by 41 minutes, and the dashboard was empty when the CEO opened it at 06:15.

Nothing failed. Every task succeeded. The backfill was correct, the nightly run was correct, and the outcome was a missed SLA with a green Airflow UI — which is Chapter 23 §23.3's distinction appearing in the orchestrator.

Three fixes, and you want all three:

• A dedicated pool for backfills, sized well below the worker count — Kestrel uses 4 of 32.
• max_active_runs=1 on anything that writes a shared target.
• A priority weight, so that when both are queued the nightly run is dispatched first.

And the operational one: an SLA on the nightly DAG's completion time, not on its tasks. Airflow's task-level SLA misses would not have fired here, because no individual task was slow — the DAG was slow because it started late, and that is a different measurement. Chapter 25 §25.5.

24.9 XCom, and What Not to Put in It

XCom passes small values between tasks, through the metadata database.

@task
def extract() -> str:
    path = write_to_s3(...)
    return path                # ← an XCom. A path. Good.

@task
def load(path: str):           # ← receives it
    ...

⚠️ Failure Mode — a DataFrame in XCom

python @task def extract() -> pd.DataFrame: return pd.read_parquet(...) # 400 MB, into the metadata database

XCom values are serialized into Airflow's metadata database, which is a PostgreSQL instance sized for scheduling metadata and shared by every DAG in the installation.

What happens, in order: the database grows · the scheduler slows for everyone · queries against the XCom table time out · and the failure presents as Airflow being broken, not as your DAG being wrong.

The rule: XCom carries references, not data. A path, an ID, a row count, a partition key. Anything you would not put in a log line does not go in XCom.

The correct pattern is the one every step of this book already uses: write to storage, pass the path. It is also what makes the task idempotent and independently retryable, because the intermediate result survives the task that produced it.

Custom XCom backends (writing to S3 transparently) exist and are a reasonable escape hatch for a team that has already built the wrong thing. They are not a reason to build it.

24.10 What Belongs in the Orchestrator

The most common structural mistake in a mature Airflow installation, and it accumulates the way Chapter 18 Case Study 1's model did — one reasonable addition at a time.

Belongs in Airflow: when things run · what depends on what · retries and alerting · concurrency and resource limits · the audit trail of what ran.

Does not: business logic · transformation SQL · data validation rules · anything you would want to unit-test.

The test: could this run correctly outside Airflow? If a task's logic only makes sense inside a PythonOperator, it cannot be tested without Airflow, cannot be run locally, and cannot be moved.

# ❌ 200 lines of pandas inside a @task.
@task
def compute_customer_segments():
    df = pd.read_sql(...)
    df["segment"] = np.where(df.ltv > 500, "high", ...)   # business logic
    ...

# ✅ The logic is a library function. The task is four lines.
from kestrel.segments import compute_segments

@task
def compute_customer_segments(data_interval_start=None):
    compute_segments(date=data_interval_start.date())

The second version is testable with pytest, runnable from a laptop, and portable to a different orchestrator. The first is none of those, and the difference is where you put a function boundary.

24.11 Testing a DAG

"You cannot test Airflow code" is a widely held belief and it is false. You cannot test it easily if your logic is inside the operators — which is §24.10's point arriving from a different direction.

Four tiers, in the order to build them.

Tier 1: the DAG imports and parses fast. Catches the majority of what breaks in practice, and runs in CI with no Airflow instance beyond the library.

# tests/test_dags.py
import time, pytest
from airflow.models import DagBag

@pytest.fixture(scope="session")
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_no_import_errors(dagbag):
    assert dagbag.import_errors == {}, dagbag.import_errors

def test_parse_is_fast():
    # §24.2: this file runs every 30 seconds on the scheduler.
    t0 = time.perf_counter()
    DagBag(dag_folder="dags/", include_examples=False)
    assert time.perf_counter() - t0 < 2.0

Tier 2: structural assertions. Cheap policy, enforced instead of remembered:

def test_every_dag_has_retries_and_an_owner(dagbag):
    for dag in dagbag.dags.values():
        assert dag.default_args.get("retries", 0) >= 1, dag.dag_id
        assert dag.default_args.get("owner"), dag.dag_id

def test_shared_writers_are_serialized(dagbag):
    for dag_id in SHARED_TARGET_DAGS:
        assert dagbag.dags[dag_id].max_active_runs == 1   # §24.8

def test_no_wall_clock(dagbag):
    # §24.3. The most consequential rule in the chapter, as a grep with
    # an allowlist -- because a comment explaining a legitimate use is
    # fine and a silent one is not.
    for path in Path("dags").rglob("*.py"):
        src = path.read_text()
        for pat in ("datetime.now(", "date.today(", "utcnow("):
            assert pat not in src or "# now-ok:" in src, (path, pat)

Tier 3: the logic, as ordinary pytest. This is the tier §24.10 exists to make possible: the transformation is a library function, tested with fixtures, with no Airflow anywhere.

Tier 4: airflow tasks test, twice. §24.5's 🔁 callout. It runs one task against a real interval without touching the metadata database, so it is the only tier that exercises the real operator, the real connection, and the real target — and running it twice is the idempotency check.

🧪 Try It — the twenty-minute audit of a DAG you already have

Four commands, and each has a defensible answer:

bash airflow dags list-import-errors # 1. should be empty time python dags/your_dag.py # 2. should be < 1s grep -rn "now()\|today()\|utcnow()" dags/ # 3. every hit explained airflow tasks test your_dag your_task 2026-03-17 # 4. then again, and diff

In the author's experience, command 3 finds something in most repositories — usually in a task that computes a filename, which is the least alarming place for it and one of the worst.

Command 2 is the one people are surprised by. A DAG file that takes four seconds to parse is not unusual, and the cause is nearly always a database call or a large read at module level that somebody added for a good reason.

24.12 How Airflow Actually Breaks

Operating an orchestrator has its own failure modes, distinct from the pipelines it runs. All four below present as "the pipelines are broken," and none of them is.

The scheduler is not running. Nothing is queued, nothing fails, and the UI shows every DAG as simply not having run. This is the most confusing failure in the system because it produces no errors anywhere — the absence of activity looks like an absence of work.

airflow jobs check --job-type SchedulerJob --hostname "$(hostname)" --limit 1

Monitor the scheduler's heartbeat as a first-class alert, and route it somewhere different from pipeline alerts, because it means something categorically different.

The metadata database is saturated. Task logs, XComs (§24.9), DAG-run rows, and rendered-template rows accumulate forever unless something removes them. A year of a busy installation is tens of millions of rows, and the symptom is that everything gets slower with nothing failing.

airflow db clean --clean-before-timestamp 2026-03-01

Run it on a schedule — it is the maintenance job every Airflow installation needs and roughly half have.

Zombie tasks. A worker dies mid-task; the scheduler eventually notices the missing heartbeat and marks the task failed. The task may have completed its work, which is §24.5's "succeeded but reported failure," and the retry that follows is safe only if the task is idempotent.

Tasks stuck in queued. The most-asked Airflow question, and it is always a limit from §24.8: parallelism, a pool with no free slots, max_active_tasks, or — with KubernetesExecutor — a cluster that cannot schedule the pod. The UI shows that a task is queued and not why, and the answer is in the scheduler log.

🏭 From the Pipeline — separate "the orchestrator is broken" from "a pipeline is broken"

These need different alerts, different runbooks, and often different people, and the single most common operational mistake here is routing them together.

	Symptom	Who fixes it
scheduler down	nothing runs, nothing fails	platform
metadata DB slow	everything slower, nothing fails	platform
worker pool exhausted	tasks queued (§24.8)	platform or the person backfilling
a task fails	one red square	the DAG's owner

Rows one and two produce no failure notification at all, which is why they must be monitored by something outside Airflow. An alert that Airflow sends you when Airflow is down is not an alert.

The cheapest external check that covers most of it: a tiny DAG scheduled every fifteen minutes that does nothing but write a timestamp, and an external monitor that alerts if the timestamp is more than thirty minutes old. It tests the scheduler, the executor, a worker, and the metadata database, end to end, in about four lines — and it is the canary that tells you the silence is not peace.

24.13 Version Note and the Alternatives

🧭 Version Note — Airflow 2.10.5, and what Airflow 3 changes

This book pins apache-airflow 2.10.5 (requirements.txt), which is what you will most commonly meet.

• execution_date is gone, replaced by logical_date and the data_interval_* pair. Anything using execution_date or {{ ds }} semantics from a 1.x tutorial needs translating, and the meaning is what §24.3 describes either way.
• schedule_interval → schedule, which also accepts a list of Datasets.
• Datasets arrived in 2.4 and are called assets in Airflow 3. Same idea, renamed.
• Deferrable operators arrived in 2.2 and need a triggerer process running. If deferrable=True appears to do nothing, that is why.
• Airflow 3 brings a rewritten UI, a task-execution API that decouples workers from the metadata database, DAG versioning, and scheduler-side asset events. The model in this chapter survives it — intervals, idempotency, pools, and "XCom carries references" are all unchanged.

This book teaches 2.10.5 because teaching an API we cannot verify would violate the numbers rule (Chapter 1). Where 3 differs materially, the section says so.

Three alternatives worth knowing, because "which orchestrator" is a real question:

Dagster is asset-oriented: you declare the data assets and their dependencies, and the graph of tasks is derived. This is the datasets idea from §24.7 taken all the way, and it is a genuinely better model for a data platform. It is younger and its ecosystem is smaller.

Prefect is Python-first with far less ceremony; a function becomes a flow with a decorator, and dynamic behaviour is easy where Airflow makes it awkward.

Managed dbt scheduling, Databricks Workflows, Snowflake tasks. For a platform that lives inside one vendor, the vendor's scheduler removes a system. The cost is that it orchestrates that vendor's things, and every data platform eventually has a step somewhere else.

Airflow's advantage is not technical. It is that it is everywhere: your next engineer has used it, every tool integrates with it, and every failure you will have has been had publicly by somebody else. That is worth a great deal, and it is the honest reason it is the default.

🎓 Interview Angle — "walk me through how you'd schedule this pipeline"

A design question disguised as a tooling question, and the tooling half is the smaller one.

The weak answer describes a DAG: extract, transform, load, in order, daily at 2 a.m.

The strong answer starts from the deadline and the dependencies:

"I'd start from the SLA — what time does someone need the number, and what happens if it's late. At Kestrel that's 06:00, and the build takes about 3.8 hours, so there's roughly 72 minutes of margin and I'd alert on the margin shrinking rather than on the breach. Every task keyed on the interval rather than the wall clock, so a backfill and a scheduled run are the same operation. max_active_runs=1 anywhere two runs would write the same target, because that's the lock. And I'd test every task by running it twice against the same interval and diffing the output — retries are configured, so any task that isn't idempotent will corrupt data on its first retry."

Four things that answer does that a DAG description does not. It starts from the consumer's deadline. It mentions the margin, which almost nobody does and which is the forward-looking number. It names max_active_runs as a correctness control rather than a throughput one. And it ends with the run-twice test, which is the thing that separates people who have been paged.

The follow-ups:

"Why not cron?" — dependencies, retries, backfills, overlap prevention, and history. And the honest concession that below about ten scheduled jobs, cron plus a lock file is genuinely simpler — a candidate who cannot say when the orchestrator is overkill has adopted it rather than chosen it.

"What is the data interval?" — the period the run covers, not the moment it fires. The payoff — a backfill and a scheduled run become the same operation — is what full marks require.

"A task failed at 3 a.m. What happens?" — retries with backoff, then an alert, then the runbook. And the good answer notes that the right question is often "can this wait until morning", which is Chapter 26's argument arriving early.

And "how do you know the scheduler is running?" — the canary, with an external monitor. This is the absence question and most candidates have never considered it, because nothing in an orchestrator's UI can tell you the orchestrator is down.

🔐 Privacy & Governance — the orchestrator sees everything and logs most of it

An orchestrator is the one system that touches every pipeline, which makes it the one system whose logs contain a sample of every dataset you own.

text where personal data ends up in an orchestrator ───────────────────────────────────────────────────────────────────── task logs anything printed. A `print(df.head())` left in from debugging is twenty rows, retained. XCom small values, in the METADATA DATABASE -- which has a different backup and a different audit trail from the warehouse rendered templates a SQL statement with a literal in it, stored per task instance and displayed in the UI connection extras a JSON blob that people put things in the UI everyone with access to the orchestrator can read every task's logs, for every pipeline

The last row is the finding. Orchestrator access is usually granted broadly — it is an operational tool — and it is effectively read access to a sample of every dataset in the platform. Almost nobody models it that way in an access review (Chapter 30).

Four controls, in order of how cheap they are:

Never log a DataFrame. A lint rule for print(df and .head() in a DAG or an operator costs nothing and removes the most common case.

Set log retention explicitly, and make it shorter than the warehouse's. airflow db clean and the remote log handler's lifecycle both need a number (Exercise 24.24), and the default is usually "forever."

Keep XCom for pointers, not payloads (§24.9). A path is fine; a row is not, and the metadata database is not a place you want personal data because it is the one store whose backups nobody classified.

And review orchestrator access as data access. In the quarterly review (Chapter 30), the orchestrator's user list belongs in the same table as the warehouse's — because functionally it is the same grant, made through a different door.

📏 Scale Note — what breaks as the DAG count grows

An Airflow deployment degrades in a specific order, and knowing it tells you which number to watch.

text DAGs tasks/day what starts hurting ───────────────────────────────────────────────────────────────────────── < 20 < 500 nothing 20-100 ~2,000 DAG parse time. Every file is re-parsed on a loop; a slow import in one DAG slows the SCHEDULER. 100-500 ~10,000 the metadata database. task_instance and log grow to tens of GB (Exercise 24.24); queries the UI runs get slow. 500+ ~50,000 the scheduler itself, and you need multiple schedulers, a partitioned metadata DB, or a different tool.

The parse-time row is the one that surprises people, because it is not about volume. Airflow re-parses every DAG file periodically; a top-level requests.get() or a database query in a DAG file runs on every parse, for every DAG, forever. One badly-written DAG can make the whole deployment sluggish, and the symptom appears on other people's DAGs.

```python

THE bug, and it is common

config = requests.get("https://config.internal/dags.json").json() # top level! for name in config["dags"]: ...

the fix: nothing expensive at import time

@task def load_config(): return requests.get(...).json() ```

The metadata-database row is Exercise 24.24's, and the fix is airflow db clean on a schedule plus a disk alert at 70%. Neither is on by default and both are two lines.

The number to watch is not the DAG count — it is dag_processing.total_parse_time, which every deployment exports and almost nobody alerts on. Above about 30 seconds, every developer's iteration loop and every schedule's punctuality are affected, and the cause is always one or two files.

🔎 Read the Plan — four views that answer "why was this late"

The UI has a lot of pages and four of them answer almost every scheduling question.

text view what it answers what people use instead ──────────────────────────────────────────────────────────────────────── Gantt where the TIME went within a run the graph view, which shows structure and no duration Landing Times how long after the interval's END the run duration, which the run FINISHED, over time hides queueing Task Duration one task's duration, over time yesterday's number Grid + queued state whether a task WAITED for a slot "it was slow"

Landing Times is the one to learn. Duration measures the run; landing time measures the promise — it is the interval's end to the run's finish, which is what the 6am SLA is actually about. A run whose duration is flat and whose landing time is climbing is queueing, and the duration chart cannot show that.

The Gantt view answers "which task" in about five seconds, and the shape tells you the class of problem: one long bar is a slow task; a staircase of short bars with gaps is a scheduler that is behind; several bars starting at once and finishing at once is a pool at its limit.

And the queued state on the grid is the one people miss entirely. A task that spent forty minutes queued and four minutes running has a concurrency problem, and every minute spent optimising the task itself is wasted.

sql -- the same question, from the metadata database, if you prefer numbers SELECT task_id, avg(extract(epoch from (start_date - queued_dttm))) AS avg_queue_s, avg(extract(epoch from (end_date - start_date))) AS avg_run_s FROM task_instance WHERE dag_id = 'kestrel_daily' AND start_date > now() - interval '30 days' GROUP BY 1 ORDER BY 2 DESC;

avg_queue_s approaching avg_run_s is a pool or a parallelism limit, not a slow task — the same comparison as Chapter 8 §8.9's warehouse queueing, one layer up.

🧭 Version Note — Airflow 3, and what a 2.x habit costs you

This book pins Airflow 2.10.5 and Airflow 3 is a real break. The concepts survive; several spellings do not.

text 2.x 3.x ───────────────────────────────────────────────────────────────────────── execution_date logical_date -- and the old name is gone, not deprecated `schedule_interval=` `schedule=` Datasets Assets, with watchers and richer conditions SubDAGs removed. Use TaskGroups. a task can reach the a task API boundary: tasks run remotely and metadata DB directly talk over an API. This is the big one. SLA misses reworked; the old SLA callback is gone

The task API boundary is the change with consequences beyond a rename. In 2.x a task could open a session against the metadata database — and plenty of production code does, to look up a previous run or to write state. In 3.x it cannot, and that code has no straightforward port.

The habit worth adopting now, on 2.x, is the one that makes the upgrade cheap:

Never touch the metadata database from a task. If you need state, use XCom for pointers or a table you own (Exercise 25.23a's run records). A platform that keeps its own run-record table is a platform whose operational history survives an orchestrator upgrade — which is a much better reason for the table than the one Exercise 25.23 gives.

And keep DAGs thin (§24.7). Business logic inside an operator is business logic that has to be ported; the same logic in an importable package is a call site that changes.

What does not change at all: the interval model, catchup, the four concurrency limits, and every failure mode in §24.8. The date model in particular is unchanged in substance and renamed in spelling, which is the most confusing possible combination and is worth knowing before you read a mixed-version answer online.

🧪 Try It — the run-twice test, on every task

bash for task in $(airflow tasks list kestrel_daily); do echo "=== $task" airflow tasks test kestrel_daily "$task" 2026-11-27 snapshot_target "$task" > /tmp/run1.txt airflow tasks test kestrel_daily "$task" 2026-11-27 snapshot_target "$task" > /tmp/run2.txt diff /tmp/run1.txt /tmp/run2.txt && echo " IDEMPOTENT" || echo " *** NOT ***" done

airflow tasks test runs a single task against a given interval without a scheduler, without a database record, and without triggering anything downstream — which is exactly what you want, and is a facility most people do not know exists.

What to expect, from every cohort that has run this: one or two tasks fail. The usual causes, in order:

An append-mode write. The second run doubles the rows.

A wall-clock read. date.today() rather than the interval — so the task loaded the wrong day both times, identically, which the diff does not catch. Check the date in the output as well as the diff.

And a CREATE TABLE without IF NOT EXISTS, which fails on the second run — the least dangerous of the three, because it is loud.

Any task whose second run changes the output will corrupt data on its first retry, and retries are configured. That sentence is the whole exercise, and it is why Exercise 24.23(e) calls it the one that carries the chapter.

24.14 The Kestrel Platform

🧱 Kestrel Platform — Increment 24: kestrel_daily and kestrel_hourly

text platform/orchestration/ dags/ kestrel_daily.py ← extract → bronze → dbt build → export kestrel_hourly.py ← clickstream micro-batch kestrel_quality.py ← source freshness (Ch. 19 §19.8), 02:45 kestrel_maintenance.py ← OPTIMIZE, VACUUM, rebind, quarantine replay plugins/callbacks.py ← on_failure_callback → the runbook link tests/test_dags.py

Eight things this increment must get right:

1. No task uses datetime.now(). A test asserts it. §24.3.
2. Every DAG file parses in under a second, asserted in CI. §24.2.
3. kestrel_quality runs at 02:45, BEFORE the build, as its own DAG — Chapter 19 §19.8's freshness check, scheduled separately so a stale source stops the build rather than being discovered inside it.
4. A warehouse_writes pool of 1, and max_active_runs=1 on every DAG writing a shared target. Chapter 20 §20.4's lock.
5. A backfill pool of 4 out of 32 worker slots, with lower priority than the nightly run.
6. Every cross-DAG dependency is a Dataset, not a sensor. Any sensor that remains is deferrable and has a comment saying why it is not a dataset.
7. XCom carries paths and counts only. A test asserts no task returns anything larger than a kilobyte.
8. Every @task body is under fifteen lines and calls into kestrel/, which is importable and testable without Airflow. §24.10.

The exercise that matters is 24.23(e): run airflow tasks test twice on the same interval for every task in kestrel_daily, and diff the target after each. Any task whose second run changes the output is a task that will corrupt data on its first retry — and retries are configured, so it will.

24.15 Summary

Cron is right until the fourth dependent step, or until a step's duration varies enough that a fixed offset is a guess. An orchestrator costs a scheduler, a database, a web server, workers, and a new failure surface — Kestrel's managed Airflow is $310 a month before any task runs.

DAG, task, operator, executor. The TaskFlow API derives the graph from the call structure — the same idea as dbt's ref(): the graph comes from the code rather than beside it.

⚠️ A DAG file is executed every parse, every thirty seconds, on the scheduler. Import, define, nothing else. time python your_dag.py should be under a second.

Airflow schedules intervals, not instants. A run labelled 2026-03-17 executes on the 18th and produces the 17th's numbers. The design is confusing and correct.

⚠️ datetime.now() in a task is the most consequential line in this chapter. A retry processes a different window; a rerun writes today's data into last March's partition; and every idempotency guarantee from Chapter 20 is defeated. data_interval_start/_end are stable forever. Grep for now().

catchup=True with a two-year-old start_date schedules 730 runs on deploy. Right for independent time windows; wrong for anything maintaining a current state.

🔁 Retries make non-idempotency a certainty, not a risk. A write that succeeded and lost its acknowledgment is the normal failure mode of a network. airflow tasks test twice, then diff the target.

A task should be the smallest unit you would want to retry independently — the unit of failure is the unit of design. Task groups organize without adding boundaries; dynamic task mapping is the supported way to fan out over a runtime value.

💸 A sensor holds a worker slot while doing nothing. Sixteen sensors × 41 minutes = 656 slot-minutes a night out of 32 slots, and it presented as the DAG being slow. deferrable=True is one keyword. Better still, a cross-DAG sensor is a dependency written as a poll — use a dataset.

⚠️ Pools are Chapter 20's lock. A backfill in the default 128-slot pool starved the nightly run and missed the 6am SLA by 41 minutes with every task green. Dedicated pool, max_active_runs=1, priority weight — and an SLA on the DAG's completion, because no individual task was slow.

⚠️ XCom carries references, not data. A DataFrame in XCom goes into the shared metadata database and presents as Airflow being broken. Write to storage, pass the path — which is also what makes the task independently retryable.

Keep business logic out of the orchestrator. The test: could this run correctly outside Airflow? A @task should be four lines calling a library function.

DAGs are testable in four tiers: import and parse time · structural policy (retries, owners, max_active_runs, no wall clock) · the logic as ordinary pytest · and airflow tasks test twice. 🧪 Four commands audit a DAG you already have in twenty minutes, and command 3 — grepping for now() — finds something in most repositories.

🏭 "The orchestrator is broken" and "a pipeline is broken" need different alerts and different people. A dead scheduler produces no failures anywhere — the absence of activity looks like an absence of work — and an alert Airflow sends you when Airflow is down is not an alert. A four-line canary DAG plus an external monitor covers the scheduler, the executor, a worker, and the metadata database.

Chapter 25 takes the thing this chapter kept saying — every task was green and the outcome was wrong — and asks what you would have to measure instead.

Key terms: DAG · task instance · operator · executor · DagBag · zombie task · db clean · canary DAG · data interval · logical date · catchup · backfill · sensor · deferrable operator · triggerer · pool · max_active_runs · task group · dynamic task mapping · XCom · dataset · clearing · TaskFlow API