Appendix D: Airflow Reference

Pinned to Apache Airflow 2.10.5 (Appendix A). §D.11 covers what changes in Airflow 3.

The concepts matter more than the API, and §D.1 is the one that produces the most bugs.

D.1 Logical Date, Data Interval, and Wall Clock

Three different times, and confusing them is the most common Airflow bug.

a DAG with schedule="0 2 * * *" (02:00 daily), run on 15 November

data_interval_start   2026-11-14 00:00   the data this run is responsible for
data_interval_end     2026-11-15 00:00
logical_date          2026-11-14 00:00   == data_interval_start
wall clock            2026-11-15 02:00   when it actually executed

The run at 02:00 on the 15th processes the 14th. That is deliberate: a daily interval is not complete until it ends.

Every task must key on the interval, never on now():

@task
def load(data_interval_start=None, data_interval_end=None):
    # correct: a rerun of this run processes the same window
    extract(since=data_interval_start, until=data_interval_end)

@task
def load_wrong():
    extract(since=datetime.now() - timedelta(days=1))   # a rerun gets different data

This is what makes a rerun idempotent (ch20, ch24), and it is why "clear and rerun" is safe.

D.2 A DAG, Current Style

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

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

@dag(
    dag_id="kestrel_daily",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5),
        "retry_exponential_backoff": True,
        "max_retry_delay": timedelta(minutes=30),
        "owner": "data-platform",
    },
    tags=["kestrel", "daily"],
    doc_md=__doc__,
)
def kestrel_daily():

    @task(outlets=[ORDERS])
    def extract_orders(data_interval_start=None, data_interval_end=None):
        ...

    @task
    def transform():
        ...

    @task(trigger_rule="none_failed_min_one_success")
    def publish():
        ...

    extract_orders() >> transform() >> publish()

kestrel_daily()

D.3 The Settings That Matter

@task(execution_timeout=timedelta(hours=2), pool="warehouse", pool_slots=2)
def heavy(): ...

D.4 Sensors, and the Deadlock

A sensor waits for a condition and holds a worker slot while it does.

# wrong: 24 sensors x 6 hours will exhaust the pool and deadlock the DAG
wait = S3KeySensor(task_id="wait", bucket_key="s3://bronze/orders/_SUCCESS",
                   poke_interval=60, timeout=60 * 60 * 6)

# right: releases the slot while waiting
wait = S3KeySensor(task_id="wait", bucket_key="s3://bronze/orders/_SUCCESS",
                   poke_interval=60, timeout=60 * 60 * 6,
                   deferrable=True)                 # or mode="reschedule"

deferrable=True uses the triggerer process and is the modern answer. mode="reschedule" releases the slot between pokes and is available everywhere.

Always set a timeout. A sensor without one waits forever, and "forever" includes the day the upstream is decommissioned.

D.5 Datasets: Scheduling on Data

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

@dag(schedule="0 2 * * *")
def producer():
    @task(outlets=[ORDERS])
    def load(): ...
    load()

@dag(schedule=[ORDERS])                 # runs when ORDERS is updated
def consumer():
    @task
    def transform(): ...
    transform()

Two DAGs, no cron guessing, no ExternalTaskSensor. The consumer runs when the producer says the data is there.

Multiple datasets are ANDed: schedule=[ORDERS, CUSTOMERS] waits for both.

D.6 Dynamic Task Mapping

@task
def list_partitions(data_interval_start=None):
    return [f"{data_interval_start.date()}/{h:02d}" for h in range(24)]

@task(max_active_tis_per_dag=4)
def load_partition(part: str):
    ...

load_partition.expand(part=list_partitions())

Useful, and it makes the DAG's shape depend on data, which complicates monitoring — a run with zero mapped tasks succeeds silently.

Assert the count:

@task
def list_partitions(...):
    parts = [...]
    if not parts:
        raise ValueError("no partitions to load -- upstream is empty")
    return parts

D.7 Trigger Rules

The common mistake: a task after a BranchPythonOperator with the default all_success is skipped, because one branch was skipped. Use none_failed_min_one_success.

D.8 XCom: What It Is Not

XCom passes small values between tasks through the metadata database.

@task
def a(): return {"rows": 4193, "path": "s3://bronze/orders/2026-11-14/"}

@task
def b(info: dict): print(info["rows"])

b(a())

It is not a data transport. Passing a DataFrame through XCom writes it into the Airflow database, which is a way to make the scheduler slow for everybody. Pass a path; read the data in the task.

D.9 Idempotency Patterns

# 1. delete-and-insert by partition -- the simplest correct pattern
@task
def load(data_interval_start=None):
    part = data_interval_start.strftime("%Y-%m-%d")
    con.execute("DELETE FROM bronze.orders WHERE event_date = ?", [part])
    con.execute("INSERT INTO bronze.orders SELECT * FROM read_parquet(?)",
                [f"s3://bronze/orders/event_date={part}/*.parquet"])

# 2. write to a temp location, then atomically swap
@task
def load_atomic(data_interval_start=None):
    tmp = f"s3://scratch/{ti.run_id}/"
    write(tmp)
    promote(tmp, final)          # a single metadata commit in a table format

# 3. a run-scoped marker, so a partial rerun is detectable
@task
def finalize(data_interval_start=None):
    write_marker(f"_SUCCESS_{data_interval_start.date()}")

Pattern 1 covers most cases and requires the partition to align with the interval (ch20).

D.10 Operational Commands

airflow dags list
airflow dags list-import-errors            # the one the UI hides
airflow dags test kestrel_daily 2026-11-14 # run it locally, no scheduler
airflow tasks test kestrel_daily load 2026-11-14

airflow dags backfill kestrel_daily \
  --start-date 2026-11-01 --end-date 2026-11-14 --reset-dagruns

airflow tasks clear kestrel_daily --start-date 2026-11-14 --end-date 2026-11-14 --downstream

airflow pools list
airflow pools set warehouse 8 "warehouse concurrency"

airflow db check
airflow db clean --clean-before-timestamp 2026-01-01   # the metadata DB grows

airflow dags list-import-errors is the first thing to run when a DAG does not appear. The UI shows nothing; this shows the exception.

D.11 🧭 Airflow 3

Airflow 3 renames and restructures several things this appendix teaches. The concepts survive; the names do not.

The task-to-database change is the consequential one: in 3.x a task cannot query the Airflow metadata database directly, which breaks a common (and always ill-advised) pattern of reading task state from inside a task.

What does not change: the logical date, the data interval, idempotency, sensors holding slots, trigger rules, and the reason catchup=True surprised you.

D.12 When Not to Use Airflow

Airflow is a scheduler with dependencies. It is not:

A data transport. Do not move data through it; move it between systems and have Airflow say when.

A streaming system. Its minimum granularity is a schedule, and Chapter 29's four questions usually say batch anyway.

A necessity. For fewer than about ten jobs with simple dependencies, cron plus a lock file plus a Slack webhook is genuinely sufficient and has no operational surface. Adopt an orchestrator when you have dependencies, backfills, and retries to reason about — which is Chapter 24's threshold, not a vendor's.