Lab 05 — Schedule the DAG and run a backfill

Lesson 05 · Data Engineering with Python · ~45 min

Goal: end this lab with the gharchive_hourly DAG running on an hourly schedule, six historical hours backfilled, and output files on disk for each processed interval. You will observe Airflow's catchup behaviour, inspect per-run context, and intentionally trigger a re-run to expose the silent bug that Lesson 06 will fix.

Prerequisites

The Lesson 04 lab must be complete: the Airflow Docker stack is running, the gharchive_hourly DAG appears in the UI, and at least one manual run has succeeded. If the stack is not running, start it:

cd data-engineering/sample-pipeline/airflow
docker compose up -d
open http://localhost:8080  # admin / admin

Step 1 — Update the DAG

Open data-engineering/sample-pipeline/airflow/dags/gharchive_hourly.py in your editor. Replace the entire file with the following:

"""gharchive_hourly — Module 2 Lesson 05 sample DAG.

Hourly scheduled, data-interval-aware. Each run downloads and counts
its own hour. Backfill-friendly: replaying an interval re-processes
the same hour deterministically.
"""

from __future__ import annotations

import sys
import pathlib
from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator

# Make the sample-pipeline package importable from inside the Airflow container.
# /opt/airflow/sample-pipeline is the mount point set by docker-compose.yaml.
_SAMPLE_PIPELINE_ROOT = pathlib.Path("/opt/airflow/sample-pipeline")
if _SAMPLE_PIPELINE_ROOT.exists() and str(_SAMPLE_PIPELINE_ROOT) not in sys.path:
    sys.path.insert(0, str(_SAMPLE_PIPELINE_ROOT))


def _hour_key(context) -> str:
    """Format the data interval start as YYYY-MM-DD-H (gharchive's 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 change: was None
    catchup=True,                        # L05 change: backfill all missed hours
    max_active_runs=2,                   # rate-limit gharchive.org politely
    tags=["data-engineering", "module-2"],
    default_args={"owner": "data-eng"},
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=_extract)
    transform = PythonOperator(task_id="transform", python_callable=_transform)
    extract >> transform

Walk through the three key changes:

• _hour_key(context) — extracts the gharchive hour string from the Airflow run context. Both _extract and _transform call it, so neither has a hardcoded date.
• schedule="@hourly" — the DAG now fires automatically at the top of every hour. Previously it was None (manual trigger only).
• catchup=True — Airflow will queue every missed hourly interval from start_date (2026-05-01) to now. max_active_runs=2 ensures at most two hours run concurrently.

Step 2 — Reload and unpause

Airflow re-parses every DAG file every 30 seconds. You do not need to restart the stack. Open http://localhost:8080 and navigate to the DAGs list. The gharchive_hourly row should still appear.

The DAG is paused by default when its schedule changes. Toggle the pause switch off — the slider at the far left of the DAG row. The scheduler will now start queuing catchup runs.

Step 3 — Watch catchup happen

Click the DAG name to open its detail page, then click Grid. Within about 30–60 seconds you should see columns appearing — one per DagRun. Each column represents one hourly interval being processed, starting from 2026-05-01T00:00:00.

With max_active_runs=2, at most two columns are in the running state (green border, animated) at the same time. The rest queue as scheduled (dark grey) and work their way through as earlier runs complete.

Let a handful of runs complete before proceeding — you only need 3–5 successes to continue. You can pause the DAG again (re-toggle the slider) if you want to stop the catchup at a specific point.

Step 4 — Inspect a single run

In Grid view, click any green cell in the extract row. A panel opens on the right. Click Logs. Find the line that shows which hour was fetched — it will look like:

[2026-05-26 09:15:42] INFO - fetching hour=2026-05-01-7 ...

The hour in the log should match the data interval shown in the Grid cell header. If it does, your DAG is correctly reading the hour from data_interval_start. If the log shows today's date instead, there is a hardcoded datetime.utcnow() somewhere — go back and fix it.

Step 5 — Pause and trigger a targeted backfill

Pause the DAG (slider off). Then run a targeted backfill for 6 specific hours using the CLI:

docker compose exec airflow-scheduler airflow dags backfill gharchive_hourly \
    --start-date 2026-05-02 \
    --end-date 2026-05-02T05:00:00

This enqueues 6 runs covering hours 0–5 of 2026-05-02. Switch to Grid view and watch the backfill__-prefixed runs flow through. They appear in a separate cluster from the scheduled runs.

Step 6 — Verify output on disk

On your host machine (outside the container), list the output files:

ls data-engineering/sample-pipeline/data/out/ | head -20

You should see one .csv file per processed hour (e.g., 2026-05-01-0.csv, 2026-05-01-1.csv, etc.). Run a quick summary with DuckDB if you have it installed:

duckdb -c "
SELECT
    regexp_extract(filename, '(\d{4}-\d{2}-\d{2}-\d+)', 1) AS hour,
    COUNT(*) AS event_types
FROM read_csv(
    'data-engineering/sample-pipeline/data/out/*.csv',
    filename=true
)
GROUP BY 1
ORDER BY 1
LIMIT 20"

You should see one row per processed hour, with the count of distinct event types.

Step 7 — Re-run an interval that already ran

Trigger a backfill for an interval that already has a successful run:

docker compose exec airflow-scheduler airflow dags backfill gharchive_hourly \
    --start-date 2026-05-01 \
    --end-date 2026-05-01T01:00:00 \
    --reset-dagruns

The DAG re-runs hour 0 of 2026-05-01. The output file data/out/2026-05-01-0.csv is overwritten in place. The run succeeds — but notice this is a silent bug: if the second run had failed halfway through, you would have a corrupt or partial file that looks complete. Module 1's transform does not deduplicate or verify atomicity. Lesson 06 addresses this.

What you should leave with

• A scheduled, backfilling gharchive_hourly DAG reading its hour from data_interval_start
• Comfort reading Grid view — identifying which interval each column covers
• Experience running a targeted CLI backfill and watching it queue DagRuns
• Awareness that "the same interval run twice" is not yet handled safely — that's Lesson 06

Troubleshooting