Scheduling & Backfills

Lesson 05 · Data Engineering with Python · ~12 min read

1. The two clocks

Every Airflow run has two timestamps, and confusing them is the single most common mistake engineers make when first working with Airflow. The first timestamp is the logical date, also called the data interval start. It is the start of the time window whose data this run is responsible for processing. If the DAG is scheduled hourly and the window is 12:00–13:00, the logical date is 12:00, regardless of when the run actually fires.

The second timestamp is the physical execution time: the wall-clock time at which the run actually started. For a live scheduled run, the two are close together — the hourly run for 12:00–13:00 fires at roughly 13:00. But during a backfill, the physical execution time is today, even though the logical date might be six months ago. A task that reads datetime.utcnow() to figure out what data to fetch will fetch today's data during the backfill run for last November — producing wrong output that silently looks correct.

The rule is simple: always derive the data window from context["data_interval_start"] and context["data_interval_end"]. Never use datetime.utcnow() or any other wall-clock reading inside a task callable. The logical date is fixed for the lifetime of a DagRun; the physical execution time is not.

This distinction matters for retries too, not just backfills. If a task fails and Airflow retries it five minutes later, data_interval_start is identical in both attempts — but datetime.utcnow() would return a different value. A task that uses the logical date is therefore deterministic: every attempt for the same run produces the same result.

2. Schedule expressions

Airflow accepts several formats for the schedule parameter. The simplest are the preset strings: "@hourly", "@daily", "@weekly", and "@monthly". These map to standard cron patterns at fixed UTC boundaries. For more control, use a cron string directly: "0 6 * * *" for 06:00 UTC every day, or "15 * * * *" for 15 minutes past every hour. You can also pass a Python timedelta object for intervals that are not aligned to clock boundaries.

One important nuance: "@daily" runs at midnight UTC. If your start_date is datetime(2026, 5, 1), the first scheduled run fires at 2026-05-02T00:00:00Z, and it covers the interval 2026-05-01T00:00:00Z to 2026-05-02T00:00:00Z. This is correct behaviour — the run fires after the interval ends — but it surprises engineers who expect the first run to fire immediately at start_date.

3. Catchup behaviour

The catchup parameter controls what Airflow does when the scheduler first encounters a DAG (or when you turn on a previously-paused DAG) and there are past intervals between start_date and now that have never been run.

With catchup=True (the Airflow default), the scheduler queues a DagRun for every missed interval from start_date to now. Consider a daily DAG with start_date=datetime(2026, 1, 1) that you first turn on on 2026-05-26: Airflow will queue 146 runs (one per day from January 1 to May 25). For an hourly DAG with the same start_date, that is approximately 3,480 runs.

With catchup=False, Airflow skips all the missed intervals and starts from the most recent one. The same daily DAG turned on 2026-05-26 gets exactly one queued run (for 2026-05-25). This is the right choice for alerting or monitoring pipelines where old intervals are irrelevant.

Choose deliberately. The safest approach for a new pipeline is to start with catchup=False and run backfills manually with the airflow dags backfill CLI command when you need history. This gives you full control over the rate and ordering of historical runs.

4. The data-interval-aware DAG

The pattern for writing a data-interval-aware task is straightforward: read the interval from context["data_interval_start"] and pass it to your extract or transform function as a parameter. The key is never to derive "what to process" from a wall-clock reading inside the task body.

For the gharchive pipeline, the extract function needs the hour string in the format YYYY-MM-DD-H (gharchive's URL convention, with no zero-padding on the hour). The correct way to derive it:

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",
    )

Because data_interval_start is a Pendulum DateTime object (Airflow uses Pendulum internally), strftime works exactly as it does on a standard Python datetime. The %-H format specifier strips the leading zero from the hour on Linux and macOS (e.g., 7 instead of 07), matching gharchive's URL convention.

Now replaying hour 7 of 2026-05-01 always calls fetch_hour(hour="2026-05-01-7"), whether the run fires today, next year, or during a manual backfill. The task is deterministic and replayable.

5. Manual backfills

The airflow dags backfill CLI command enqueues DagRuns for a specified date range. It is the correct tool when you want to re-process historical data — for example, because you deployed a new transform and need to apply it to the last month, or because a data bug was found and you need to replay the affected days.

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

This enqueues six DagRuns, covering the intervals 00:00–01:00, 01:00–02:00, ..., 05:00–06:00 of 2026-05-01. The runs execute concurrently up to max_active_runs. Because the DAG is data-interval-aware, each run independently fetches its own hour — no manual parameter changes needed.

Two practical notes. First, backfill runs are marked with a backfill__ prefix in their run_id, which distinguishes them from regularly scheduled runs in the Grid view. Second, if the DagRuns already exist (from a previous scheduled run), you must pass --reset-dagruns to allow the backfill to overwrite them. Without that flag, the command silently skips existing runs.

6. Concurrency controls

Three parameters govern how many tasks run simultaneously for a DAG.

max_active_runs (default: 16) limits the number of DagRun instances that can be in the running state at the same time for a given DAG. Setting it to 2 means at most two hourly intervals are being processed concurrently. This is the primary knob for rate-limiting an extract that calls an external API.

max_active_tasks (default: 16) limits the total number of task instances that can be running across all concurrent DagRuns for a given DAG. If you have max_active_runs=4 and each run has 3 tasks, the theoretical maximum concurrency is 12 task instances — but you can cap it lower with max_active_tasks.

Pools are named slot budgets that apply globally, across all DAGs. Create a pool called gharchive_api with 4 slots (in the Airflow UI under Admin → Pools), then wire the extract task to use it:

PythonOperator(
    task_id="extract",
    python_callable=_extract,
    pool="gharchive_api",
)

Now at most 4 extract tasks run at once globally, regardless of how many DAGs reference gharchive_api. This is the correct tool when multiple DAGs all call the same external API and you need a single throttle.

7. Common pitfalls

References: Airflow Scheduler docs · Time Zones · Datasets feature · Backfill CLI reference