Lesson 05 — Scheduling & Backfills

Schedules, catchup, and the data-interval-aware DAG pattern.

Why scheduling is harder than it looks

• Every run has a time window — e.g., "process the hour between 12:00 and 13:00"
• The schedule is just when do we run it — the data interval is what window do we process
• Mixing the two is the #1 Airflow footgun: reading datetime.utcnow() inside a task gives a different answer on every replay
• Airflow fires each run after the interval it processes has elapsed — the run for 12:00–13:00 fires at 13:00, not 12:00

The two clocks

Schedule expressions

Catchup, explained

When you set start_date=2026-05-01 and the scheduler starts on 2026-05-26 with catchup=True, Airflow queues every missed interval from start_date to now — approximately 600 hourly runs in that example. With catchup=False, Airflow skips straight to the most recent interval.

Choose deliberately — the default in Airflow is True.

The data-interval-aware DAG

# L04 (hardcoded — breaks on backfill):
def _extract(**context):
    from extract.gharchive import fetch_hour
    fetch_hour(hour="2026-05-01-12")

# L05 (data-interval-aware — safe to replay):
def _extract(**context):
    from extract.gharchive import fetch_hour
    hour = context["data_interval_start"].strftime("%Y-%m-%d-%-H")
    fetch_hour(hour=hour)

Now every run processes its own hour. Replaying the same interval always fetches the same data.

What's in context

• data_interval_start — pendulum DateTime; start of the interval this run processes
• data_interval_end — pendulum DateTime; end of the interval
• logical_date — alias for data_interval_start (Airflow 2.2+)
• run_id — unique identifier for this DagRun
• task_instance — access XCom, retry count, etc.
• params — JSON payload passed at trigger time

def _extract(**context):
    print(context["data_interval_start"])  # pendulum DateTime
    print(context["run_id"])               # 'scheduled__2026-05-01T12:00:00+00:00'

Triggering a backfill

airflow dags backfill gharchive_hourly \
    --start-date 2026-05-01 \
    --end-date 2026-05-01T06:00:00

This enqueues 6 DagRuns, one per hourly interval (hours 0–5). Watch them parallel up to max_active_runs.

max_active_runs and concurrency

• By default each DAG runs at most 16 instances concurrently (max_active_runs=16)
• For backfills, raise it to speed up; lower it if your extract is rate-limited by an external API
• max_active_tasks caps the total number of running task instances across all concurrent DagRuns

with DAG(
    dag_id="gharchive_hourly",
    max_active_runs=2,      # at most 2 concurrent runs — polite to gharchive.org
    max_active_tasks=8,
    ...
) as dag: ...

The catchup gotcha

• The first time you unpause a long-backlogged DAG the scheduler queues everything at once
• External APIs often rate-limit at 2–5 req/s — 4,300 parallel requests is an outage, not a backfill
• A safe pattern: start with max_active_runs=1, watch the queue drain, then raise

Dataset triggers (one-paragraph teaser)

Airflow 2.4+ added Datasets: declare that a DAG produces Dataset("s3://bucket/path") and another DAG consumes it, and the second DAG auto-triggers when the first finishes writing to that path. This replaces fragile ExternalTaskSensor patterns for cross-DAG dependencies.

from airflow.datasets import Dataset

HOURLY_OUT = Dataset("file:///opt/airflow/sample-pipeline/data/")

# Producer DAG
with DAG(..., schedule="@hourly") as hourly_dag:
    PythonOperator(..., outlets=[HOURLY_OUT])

# Consumer DAG — fires when HOURLY_OUT is updated
with DAG(..., schedule=[HOURLY_OUT]) as rollup_dag:
    ...

Not used in this course's sample pipeline, but worth knowing for production setups.

The full L05 DAG

def _hour_key(context) -> str:
    """Format data_interval_start as YYYY-MM-DD-H (gharchive filename shape)."""
    return context["data_interval_start"].strftime("%Y-%m-%d-%-H")

def _extract(**context) -> None:
    from extract.gharchive import fetch_hour
    hour = _hour_key(context)
    fetch_hour(hour=hour)

def _transform(**context) -> None:
    from transform.events import count_by_type
    hour = _hour_key(context)
    count_by_type(
        input_path=f"data/raw/{hour}.json.gz",
        output_path=f"data/out/{hour}.csv",
    )

with DAG(
    dag_id="gharchive_hourly",
    start_date=datetime(2026, 5, 1),
    schedule="@hourly",     # ← L05: was None
    catchup=True,           # ← L05: was False
    max_active_runs=2,
    tags=["data-engineering", "module-2"],
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=_extract)
    transform = PythonOperator(task_id="transform", python_callable=_transform)
    extract >> transform

Common gotchas

• Confusing logical_date with "now" — logical_date is the start of the interval being processed, not the wall clock when the task ran. A run triggered today for last week's data has logical_date = last week.
• Using datetime.utcnow() inside a task — every replay sees a different time. The output is non-deterministic. Use context["data_interval_start"] instead.
• Forgetting that catchup=False delays the first run — if you add the DAG at 09:47, the first run fires at 10:00 (after the first full interval elapses), not immediately.

Recap & what's next

• Every run has a data interval — use context["data_interval_start"], never datetime.utcnow()
• Catchup replays missed intervals — choose catchup=True/False deliberately; always pair with a sane max_active_runs
• Data-interval-aware DAGs are replayable — the same interval, run twice, downloads the same data