Lesson 06 — Idempotency & Retries

Why pipelines must be idempotent — and how to make them so.

The silent bug from Lesson 05

In the Lab 05 bonus step, re-running an interval overwrote the Parquet file. Looks fine — right? What if the second run failed halfway through the write? Now you have a half-written file: corrupt, but named like a complete one. The filesystem shows it exists; downstream reads return garbage.

Why pipelines retry

• Networks blip — gharchive.org returns a transient 503
• The Airflow executor crashes mid-task (OOM, pod eviction, host reboot)
• You deliberately backfill — the same interval runs again by design
• A human clears a failed task to let it retry after fixing a bug

At-least-once vs exactly-once

The transactional load pattern

import os
from pathlib import Path

def transactional_write(df, final_path: str):
    staging = f"{final_path}.staging.tmp"
    df.to_parquet(staging)           # write to staging
    os.replace(staging, final_path)  # atomic rename — POSIX guarantees

• os.replace() is atomic within a single filesystem — POSIX standard
• Either the new file is there (complete) or the old one is (untouched) — never a half-file
• Encode the run_id in the staging name to avoid concurrent-run collisions

Run-key-based dedupe

When you can't atomically replace a single file (e.g., appending to a partitioned table), encode the logical date into the output path. Re-runs overwrite the same partition. Other partitions stay untouched.

# Partition by hour — replay overwrites only this hour's partition
output = f"data/event_date={date}/hour={hour}/events.parquet"
df.to_parquet(output)  # second run for same (date, hour) is safe

XCom — pass metadata between tasks

def _extract(**context):
    from extract.gharchive import fetch_hour
    path = fetch_hour(hour=_hour_key(context))
    return path   # automatically pushed to XCom

def _transform(**context):
    from transform.events import count_by_type
    path = context["task_instance"].xcom_pull(task_ids="extract")
    count_by_type(input_path=path, output_path=f"data/out/{_hour_key(context)}.csv")

XCom is metadata-only — small JSON-serializable values (paths, counts, IDs). Not for DataFrames.

XCom vs side-effects

Retry configuration

from datetime import timedelta
from airflow.operators.python import PythonOperator

extract = PythonOperator(
    task_id="extract",
    python_callable=_extract,
    retries=3,
    retry_delay=timedelta(seconds=30),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(minutes=10),
)

Exponential backoff schedule: 30 s → 60 s → 120 s → capped at 10 min. After exhausting retries, task lands in failed and the DAG run is marked failed.

Retry budgets

Non-idempotent counterexamples

• Append to a file (mode='a') — replay doubles the rows. Fix: always overwrite; use staging-then-rename.
• Increment a Postgres counter — UPDATE SET count = count + 1 — replay double-counts. Fix: upsert with ON CONFLICT DO UPDATE SET count = EXCLUDED.count.
• External API that creates a resource (e.g., POST /orders with uuid4() as the ID) — retry creates a duplicate order. Fix: derive the idempotency key from the run_key, not a random UUID.

L06 sample-pipeline upgrade

Three changes to dags/gharchive_hourly.py:

1. Staging-then-rename in both tasks — extract writes to data/<hour>.parquet.staging.<run_id>.parquet, then os.replace() to data/<hour>.parquet. Transform does the same for data/out/<hour>.csv.
2. Retry budgets split by task character — extract uses EXTERNAL_API_RETRIES (3 retries, exponential backoff); transform uses PURE_COMPUTE_RETRIES (0 retries).
3. XCom carries the path — extract returns the final path string; transform pulls it via xcom_pull instead of reconstructing it.

All presets live in a new dags/_helpers.py shared by all Module 2 DAGs.

Common gotchas

• S3 multipart uploads aren't atomic — a full-object PUT is, but multipart isn't. Always complete the multipart, then CopyObject to the final key from the staging key.
• INSERTs without ON CONFLICT — second run fails on PK violation, which Airflow retries indefinitely, masking the real bug. Write upserts, not inserts.
• Side-effects you can't undo (send Slack, call a billing API) — gate them behind a first-success check, or accept that they may fire more than once and design the receiver to be idempotent.

Recap & what's next

• Every task should be idempotent — the orchestrator will re-run it
• Staging-then-rename is the cheapest pattern: POSIX atomic, zero extra infra
• XCom for metadata (paths, counts); side-effects for data — never push DataFrames through XCom