Idempotency & Retries

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

1. Why idempotency matters

Every distributed system retries. The orchestrator delivers an at-least-once guarantee: every task runs at least once, but possibly more. Transient network errors, executor crashes, pod evictions, and deliberate backfills all cause a task to run again. If your task isn't built to handle that, you'll produce silent data corruption — rows doubled in a database, files half-written, external resources duplicated.

Idempotency is the property that running a task N times with the same inputs produces the same final state as running it once. An idempotent task combined with an at-least-once orchestrator gives you exactly-once semantics on the output — which is what every data pipeline actually needs.

The most common non-idempotent operations that look safe on paper:

• Appending to a file (mode='a') — a second run doubles every row.
• Incrementing a counter in a database (UPDATE SET n = n + 1) — a retry double-counts.
• Sending an email or Slack notification — a retry sends it twice.
• Calling an external API that creates a resource with a randomly generated ID — a retry creates a duplicate resource.

In each case the fix is the same principle: make the operation state-based rather than event-based. Instead of "append row", write "set the row identified by this key to these values". Instead of "send notification", check "did I already send for this run key?".

2. The transactional-write pattern

The simplest technique for making file writes idempotent is staging-then-rename. Write output to a uniquely named staging file, then atomically rename it to the final path. The POSIX standard guarantees that rename() (and Python's os.replace()) is atomic within a single filesystem. Either the final file is fully written (complete) or the previous version remains untouched — never a half-written intermediate state.

import os
import pathlib

def _extract(**context):
    from extract.gharchive import fetch_hour
    hour = _hour_key(context)
    final = pathlib.Path(f"/opt/airflow/sample-pipeline/data/{hour}.parquet")
    final.parent.mkdir(parents=True, exist_ok=True)
    # Encode run_id in the staging name — concurrent runs don't collide
    safe_run = context["run_id"].replace(":", "_").replace("+", "_")
    staging = final.with_suffix(f".staging.{safe_run}.parquet")
    fetch_hour(hour=hour, out_dir=str(staging.parent))  # write to staging
    os.replace(staging, final)                           # atomic rename
    return str(final)   # push final path to XCom

The same technique works for S3: upload to a staging key (e.g., s3://bucket/staging/run_id/hour.parquet), then issue a server-side CopyObject to the final key. A full-object S3 PUT is atomic; multipart uploads are not by default, so you must use the staging-then-copy pattern for large files on S3. For data warehouses, the equivalent is: write to a temp table, then INSERT OVERWRITE or MERGE INTO the target within a transaction.

One important note: encode the run_id in the staging file name. If two concurrent runs process the same hour (possible during backfills with max_active_runs > 1), they each need their own staging file. Without this encoding, one run's staging write would overwrite the other's, and the winner is non-deterministic.

3. Run-key partitioning

When you can't atomically replace a single file — for example, when you're writing to a partitioned dataset where each partition is a separate object — encode the logical date (data interval start) into the output path. Re-running the same interval then overwrites exactly that partition and leaves all other partitions untouched.

day   = context["data_interval_start"].strftime("%Y-%m-%d")
hour  = context["data_interval_start"].strftime("%H")
# Partition key is part of the path — replay is safe
output = f"data/event_date={day}/hour={hour}/events.parquet"
df.to_parquet(output)

This is the standard lakehouse pattern, formalized by table formats like Apache Iceberg and Delta Lake. When a run retries, it re-writes the same partition. Other days and hours are untouched. Module 4 builds on this pattern when it introduces Iceberg table management and partition pruning.

Run-key partitioning also makes debugging straightforward: if you suspect hour 14 of May 1 has bad data, you know exactly which file to inspect or delete. There is no ambiguity about which run wrote which rows.

4. XCom vs side-effects

Airflow provides XCom (cross-communication) as a task-to-task messaging mechanism, backed by the Airflow metadata database. A task pushes a value to XCom either by returning it from the Python callable (Airflow auto-pushes the return value under the key return_value) or by calling context["task_instance"].xcom_push(key, value) explicitly. A downstream task pulls it with context["task_instance"].xcom_pull(task_ids="upstream_task_id").

XCom is designed for small, JSON-serializable metadata: a file path, a row count, a record ID, a status flag. Do not push DataFrames or large byte strings through XCom. The metadata database (Postgres by default) stores XCom values as JSONB or BLOB; large payloads will bloat the DB, slow the Airflow UI, and cause timeouts during DAG parsing.

For actual data, use side-effects: write to a known storage location (a Parquet file at a conventional path, a warehouse table partition, an S3 key) and let downstream tasks read from it by convention. XCom carries the address, not the data itself.

def _extract(**context):
    from extract.gharchive import fetch_hour
    path = fetch_hour(hour=_hour_key(context), out_dir="data/raw")
    return path   # XCom carries the path string — not the file contents

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

This separation also keeps idempotency clean: the storage side-effect (file write) is the authoritative state. XCom is ephemeral metadata that lets tasks communicate the address of that state without hard-coding paths in multiple places.

5. Retry configuration

Airflow exposes four retry parameters on every operator:

• retries — maximum number of re-attempts after the first failure (default 0).
• retry_delay — fixed wait before the first retry (a timedelta).
• retry_exponential_backoff — if True, the wait doubles each attempt: retry_delay, 2 × retry_delay, 4 × retry_delay, …
• max_retry_delay — caps the exponential growth. Without this, a long backoff can push retries past the next schedule interval.

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

With these settings the retry schedule is: first failure → wait 30 s → second failure → wait 60 s → third failure → wait 120 s → fourth failure → task lands in failed. The cap at 10 min means no retry waits longer than 10 min even if the exponential calculation says longer.

After exhausting all retries, the task instance moves to the failed state. The upstream failure propagates downstream: tasks that depend on the failed task move to upstream_failed and are skipped. The DAG run is marked failed. Alerting (on-failure callbacks, PagerDuty integrations) fires at this point.

6. Retry budgets and on-failure callbacks

The right number of retries depends on the character of the failure you're defending against. A principled approach:

• External-API tasks — fetch from gharchive.org, call a REST endpoint, read from an S3 bucket: use retries=3 with exponential backoff. These calls fail transiently (rate limits, DNS blips, provider incidents). Three retries covers most transient failure windows.
• Pure-compute tasks — transform a DataFrame, run a SQL aggregation, aggregate a Parquet file: use retries=0. These tasks are deterministic. A failure is a real bug (schema mismatch, out-of-memory, logic error). Retrying won't fix a bug; it just delays surfacing it and wastes compute.

When retries are exhausted, you want a human to know. Airflow supports an on_failure_callback parameter on operators and on the DAG itself. Pass a Python callable that fires when the task's final attempt fails. Use it to call your alerting system (Slack, PagerDuty, email). The distinction matters: you don't want an alert on every retry attempt (noise), but you do want one when the budget is exhausted (signal).

A practical pattern for the sample pipeline: set an on-failure callback on the extract task that logs a message to the Airflow task log and, in production, would call a Slack webhook. For course purposes we keep it as a log message; the plumbing is identical.

7. Common pitfalls

References: Airflow retry docs · Airflow best practices — idempotency · S3 multipart upload · Apache Iceberg ACID operations