Lab 06 — Break the DAG mid-run, replay it, prove idempotency

Lesson 06 · Data Engineering with Python · ~45 min

By the end of this lab you will have concrete evidence — matching checksums across two independent runs of the same hour — that your DAG is exactly-once even after a mid-run failure and recovery.

Prerequisites

• Lab 05 completed: the Airflow stack is running, gharchive_hourly is scheduled and has produced at least a few hourly Parquet files in data/raw/.
• Docker Desktop (or docker compose v2+) available and the stack still up. Verify with docker compose ps from data-engineering/sample-pipeline/airflow/.

Step 1 — Update the DAG and helpers

The sample-pipeline code was updated as part of the lesson scaffolding. Read through both files before running anything — understanding the code is the goal.

dags/_helpers.py — introduces three building blocks:

def transactional_write_path(final_path: pathlib.Path, run_id: str) -> pathlib.Path:
    """Build a per-run staging path next to the final file.
    Encodes run_id so concurrent runs for the same hour don't collide."""
    safe_run = run_id.replace(":", "_").replace("+", "_")
    return final_path.with_suffix(f".staging.{safe_run}.parquet")

def atomic_replace(src: pathlib.Path, dst: pathlib.Path) -> None:
    """os.replace with explicit failure if src doesn't exist."""
    if not src.exists():
        raise FileNotFoundError(f"staging file missing: {src}")
    os.replace(src, dst)

EXTERNAL_API_RETRIES = dict(
    retries=3, retry_delay=timedelta(seconds=30),
    retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=10),
)
PURE_COMPUTE_RETRIES = dict(retries=0)

dags/gharchive_hourly.py — three changes from L05:

1. Staging-then-rename in extract: fetch_hour writes the raw .json.gz to a staging directory; atomic_replace renames it to the final path. If the task dies between the write and the rename, the staging file is visible on disk (visibly incomplete) but the final path is untouched.
2. Staging-then-rename in transform: count_by_type writes to a staging CSV derived from transactional_write_path; the final rename happens only after the write completes successfully.
3. Retry budgets split by task character: extract uses **EXTERNAL_API_RETRIES (3 retries, exponential backoff); transform uses **PURE_COMPUTE_RETRIES (0 retries, because transform failures are real bugs).

XCom now carries the final raw-file path from extract to transform, so transform doesn't hard-code a path — it reads exactly what extract produced.

Step 2 — Reload Airflow

Airflow re-parses the dags/ directory every ~30 seconds. Wait 30 seconds after the files are in place, then open http://localhost:8080. Confirm there is no "DAG Import Errors" banner at the top of the page. If an error appears, check docker compose logs airflow-scheduler for the Python traceback.

Step 3 — Clean slate

To get a deterministic starting point for the checksum test, wipe the data directory:

docker compose exec airflow-scheduler \
    rm -rf /opt/airflow/sample-pipeline/data

This removes all previously produced raw and output files. The DAG will recreate everything from scratch in the steps below.

Step 4 — Run a single hour and record the checksum

Use airflow dags test to run a single DAG run in the foreground (no scheduler needed — useful for quick local testing):

docker compose exec airflow-scheduler \
    airflow dags test gharchive_hourly 2026-05-01T12:00:00

After it completes, verify the output on the host (the mount is shared):

ls data-engineering/sample-pipeline/data/raw/
# Expected: 2026-05-01-12.json.gz

ls data-engineering/sample-pipeline/data/out/
# Expected: 2026-05-01-12.csv

Record the checksum of the raw file:

md5sum data-engineering/sample-pipeline/data/raw/2026-05-01-12.json.gz

Note the hash — you will compare it after the replay in Step 7.

Step 5 — Force a mid-run failure

In the Airflow UI, trigger a manual run for hour 13 (2026-05-01T13:00:00). While the extract task is actively running (yellow/green in the Grid view), restart the scheduler container to simulate an executor crash:

docker compose restart airflow-scheduler

The running task instance will land in failed (or remain running until the scheduler heartbeat times it out — allow up to 60 seconds). Now inspect the data directory:

ls data-engineering/sample-pipeline/data/raw/

You should see a file named something like 2026-05-01-13.json.gz.staging.scheduled__2026-05-01T13_00_00_00_00.parquet but no 2026-05-01-13.json.gz. This is the point of the staging pattern: a partial download is visibly partial — named differently, easy to find and clean up — rather than silently corrupting the final file.

Step 6 — Clear the failed run and let it retry

In the Airflow UI:

1. Open the gharchive_hourly DAG.
2. In the Grid view, click the red (failed) extract cell for the hour-13 run.
3. Click Clear. In the dialog, confirm Past=No, Recursive=No (clear only this task instance, not its downstream). Click Clear.

Airflow re-queues the extract task. Wait for it to run. If the retry budget hasn't been exhausted (it hasn't — the task was failed by a crash, not by three consecutive errors), the task should succeed. Confirm:

ls data-engineering/sample-pipeline/data/raw/
# Now includes: 2026-05-01-13.json.gz
ls data-engineering/sample-pipeline/data/out/
# Now includes: 2026-05-01-13.csv

Any orphan staging file from the crashed attempt is now irrelevant — the final file is complete.

Step 7 — Prove replay is idempotent

Re-run hour 12 — the one you already produced in Step 4:

docker compose exec airflow-scheduler \
    airflow dags test gharchive_hourly 2026-05-01T12:00:00
md5sum data-engineering/sample-pipeline/data/raw/2026-05-01-12.json.gz

Compare this checksum with the one you recorded in Step 4. They should be identical. The gharchive.org file for a past hour does not change — the same bytes are downloaded every time. Combined with the staging-then-rename write pattern, the final output is byte-for-byte reproducible across any number of replays. That is exactly-once.

Step 8 — Clean up orphan staging files

After a force-killed run, orphan staging files may remain in the data directory. A safe one-liner to clean them up (deletes only files older than 1 hour, so concurrent in-progress runs are never affected):

docker compose exec airflow-scheduler \
    find /opt/airflow/sample-pipeline/data -name "*.staging.*" -mmin +60 -delete

In production you would run this as a separate Airflow DAG (a scheduled maintenance task) rather than as a manual one-liner.

What you should leave with

• Concrete evidence (matching checksums) that your DAG is replayable regardless of how many times it runs.
• Understanding of the staging-then-rename pattern and why POSIX atomic rename is the right primitive.
• Awareness that orphan staging files are a normal side-effect of interrupted runs and need a cleanup policy.
• Confidence that the retry policy split (external API vs pure compute) catches real failures without masking bugs.

Troubleshooting