Idempotency in Data Pipelines: Why It Matters

An operation is idempotent if doing it twice has the same effect as doing it once. For data pipelines this is not an academic nicety. Jobs fail and are retried, orchestrators rerun tasks, message brokers redeliver, and engineers rerun backfills by hand. A pipeline that is not idempotent turns each of those ordinary events into wrong numbers, usually silently.

Why repeats are unavoidable

Microsoft's architecture guidance on the idempotent consumer pattern lists how duplicates arise even in well-built systems (Azure Architecture Center: idempotent consumer): a producer retries because an acknowledgement was lost although the first send succeeded; a consumer processes a message but crashes before acknowledging it, so it is delivered again; a write completes but the worker dies before recording success. Most brokers promise at-least-once delivery, and the guidance is blunt that exactly-once delivery across a distributed system is impractical. The durable answer is not to prevent repeats but to make them harmless. The same guidance extends the principle beyond messaging to ETL reprocessing, streams resuming from checkpoints, and scheduled jobs that restart.

Four load strategies under stress

The test below loads a month of daily trip counts per pickup zone, built from January 2024 New York yellow taxi records (NYC TLC: trip record data), into SQLite with four strategies. Each strategy faces two ordinary events: the job runs twice, and a reload crashes halfway and is retried. The table shows total trips as a percentage of the true total:

import sqlite3
import urllib.request
from pathlib import Path
import pandas as pd

name = "yellow_tripdata_2024-01.parquet"
if not Path(name).exists():
    urllib.request.urlretrieve(f"https://d37ci6vzurychx.cloudfront.net/trip-data/{name}", name)
raw = pd.read_parquet(name, columns=["tpep_pickup_datetime", "PULocationID"])
raw = raw[raw["tpep_pickup_datetime"].between("2024-01-01", "2024-02-01", inclusive="left")]
batch = (raw.assign(day=raw["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d"))
            .groupby(["day", "PULocationID"]).size().reset_index(name="trips"))
rows = list(batch.itertuples(index=False, name=None))
TRUE_TRIPS = int(batch["trips"].sum())
print(f"batch: {len(rows)} day-zone rows, {TRUE_TRIPS} trips")

class Crash(Exception):
    pass

def crashing(rows, after):
    for i, row in enumerate(rows):
        if i == after:
            raise Crash
        yield row

def fresh():
    con = sqlite3.connect(":memory:")
    con.execute("CREATE TABLE daily (day TEXT, zone INTEGER, trips INTEGER)")
    return con

def append(con, rows):
    with con:
        con.executemany("INSERT INTO daily VALUES (?, ?, ?)", rows)

def replace_no_txn(con, rows):
    con.execute("DELETE FROM daily WHERE day LIKE '2024-01-%'"); con.commit()
    for row in rows:
        con.execute("INSERT INTO daily VALUES (?, ?, ?)", row); con.commit()

def replace_in_txn(con, rows):
    with con:
        con.execute("DELETE FROM daily WHERE day LIKE '2024-01-%'")
        con.executemany("INSERT INTO daily VALUES (?, ?, ?)", rows)

def upsert(con, rows):
    con.execute("CREATE UNIQUE INDEX IF NOT EXISTS daily_key ON daily (day, zone)")
    with con:
        con.executemany("INSERT INTO daily VALUES (?, ?, ?) ON CONFLICT (day, zone) DO UPDATE SET trips = excluded.trips", rows)

total = lambda con: con.execute("SELECT COALESCE(SUM(trips), 0) FROM daily").fetchone()[0] / TRUE_TRIPS
half = len(rows) // 2
print(f"{'strategy':<28}{'run twice':>10}{'after crash':>13}{'crash, retry':>14}")
for name, load in [("append", append), ("delete + insert, no txn", replace_no_txn),
                   ("delete + insert, one txn", replace_in_txn), ("upsert on key", upsert)]:
    twice = fresh(); load(twice, rows); load(twice, rows)
    crash = fresh(); load(crash, rows)                       # yesterday's good load
    try:
        load(crash, crashing(rows, half))
    except Crash:
        pass
    after_crash = total(crash)
    load(crash, rows)
    print(f"{name:<28}{total(twice):>9.0%}{after_crash:>13.0%}{total(crash):>14.0%}")

batch: 6949 day-zone rows, 2964606 trips
strategy                     run twice  after crash  crash, retry
append                           200%         100%          200%
delete + insert, no txn          100%          47%          100%
delete + insert, one txn         100%         100%          100%
upsert on key                    100%         100%          100%

Reading the results

Append is not idempotent. Running it twice doubled every total. Its crash case looks fine at first, because the failed batch ran inside a transaction and rolled back, but the retry appended the month again on top of the data already there: 200%. Appending is only safe when something else guarantees each row arrives once.

Replace without a transaction is idempotent but unsafe. Rerunning it gives the right answer, because it deletes the month first. But when it crashed halfway, the old rows were already deleted and committed, and only half the new ones had arrived: for as long as the failure lasted, anyone reading the table saw 47% of January's trips. The retry repaired it, but a dashboard refreshed in between would have been badly wrong.

Replace in one transaction is idempotent and safe. The delete and the inserts commit together or not at all; Python's sqlite3 connection commits a successful block and rolls back a failed one (Python docs: sqlite3). After the crash, readers still saw yesterday's complete data, and the retry replaced it cleanly.

Upsert on a key is idempotent and safe. Each row is inserted, or updated if its day and zone already exist, SQLite's UPSERT (SQLite docs: UPSERT). Repeats rewrite the same values. Microsoft's guidance recommends exactly this kind of naturally idempotent operation, an upsert on a business key or a write of absolute values, wherever possible.

Patterns that make pipelines idempotent

• Replace whole partitions atomically, a day or a month, as in building your first data pipeline.
• Upsert or merge on a natural key, never on a generated one. A random ID or a load timestamp in the key makes every run look new.
• Write absolute values, not increments. Setting a count to 42 can be repeated; adding 1 cannot.
• Record progress with the data. Save a watermark or offset in the same transaction as the load, so a crash cannot separate them; incremental loads covers this.
• Deduplicate replayed events with a stable event ID and a uniqueness constraint, as change data capture consumers must; see CDC explained.
• Make side effects idempotent too: a pipeline that emails a report or calls an API should record that it did, so a retry does not send it twice.

Idempotency in streaming

Streaming makes repeats routine rather than exceptional. A stream processor records checkpoints of how far it has read; after a failure it restarts from the last checkpoint and processes again every event received since, which Microsoft's guidance lists among the cases that need idempotent processing. The sink therefore has to absorb replays: upserts keyed on an event or entity ID, writes that commit together with the checkpoint, or deduplication on a stable event ID within a time window. A streaming job whose sink simply appends will slowly overcount, a little more after every restart.

Testing for idempotency

The table above is itself a test design worth copying. For each pipeline step, check that running it twice produces identical output, and that killing it midway and rerunning ends in the same state. Automated in continuous integration on a small fixture dataset, these two checks catch most idempotency bugs before they reach production. Where this fits in overall pipeline design is covered in ETL vs ELT.

Data loading strategies, workflow automation and logging and error handling are part of the Data Engineering course, in its ETL and Python modules.

Sources

• Idempotent Consumer pattern (Azure Architecture Center, Microsoft Learn) — Microsoft Learn
• UPSERT (SQLite documentation) — SQLite
• sqlite3: DB-API 2.0 interface for SQLite databases (Python documentation) — Python documentation
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission