Idempotent Data Pipelines: Safe Retries Without Duplicates

idempotent data pipelines are the difference between a pipeline you can retry with confidence and one you fear touching, because every re-run risks double-counting revenue, duplicating rows, or firing the same downstream side effect twice. Failures in a distributed data stack are not rare events to be engineered away — a worker gets OOM-killed mid-write, a network blip drops an acknowledgement, a Kafka consumer rebalances, an Airflow task times out and the scheduler retries it. In every one of those cases something runs a second time, and the only question that matters is whether running it a second time leaves your warehouse, your topic, and your downstream systems in exactly the same state as running it once. That property — same state after one run or many — is idempotency, and it is the load-bearing contract of any pipeline that is allowed to retry.

This guide is the walkthrough you wished existed the first time an on-call page said "the nightly aggregation ran twice and finance is asking why revenue doubled," or an interviewer asked "your consumer restarts mid-batch — how do you guarantee no duplicates?" It works through the four levers that make retries safe: the deterministic idempotency key and the dedup store that remembers it, the upsert / MERGE sink that updates by key instead of appending, the honest decomposition of exactly-once into at-least-once delivery plus idempotent processing plus checkpointing, and the design rules for retry-safe Airflow and Spark tasks keyed on a logical interval instead of the wall clock. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works. All SQL is PostgreSQL dialect, but the mental model carries to Snowflake, BigQuery, Delta Lake, and every other keyed sink.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse on the data processing practice library →, and sharpen the streaming axis with the streaming practice library →.

---

On this page

• Why idempotency is the retry contract
• Idempotency keys & dedup stores
• Upsert / MERGE & idempotent sinks
• Exactly-once vs at-least-once + checkpoints
• Designing idempotent tasks (Airflow / Spark)
• Cheat sheet — idempotent pipeline recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why idempotency is the retry contract

Retries are inevitable, so correctness has to survive them — that survival property is idempotency

The one-sentence invariant: a data-pipeline operation is idempotent when applying it once and applying it N times (for any N ≥ 1) leave the target system in the same final state — and because every distributed pipeline retries on failure, idempotency is not a nice-to-have but the contract that lets you retry at all without shipping duplicates, double-counted metrics, or repeated side effects. The instinct of a junior engineer is to make failures rare; the instinct of a senior engineer is to make retries safe. You cannot drive the failure probability of a distributed system to zero — you can only guarantee that when the inevitable retry happens, it converges to the same answer as a clean single run.

The axes that matter.

• Unit of work. Idempotency is always defined relative to a unit — a single event, a batch, a partition, a task instance for a logical interval. "Is this pipeline idempotent?" is meaningless until you name the unit. Re-processing one Kafka message must be safe; re-running one day's aggregation must be safe. Pick the unit before you reason about anything else.
• Dedup key. Every idempotent operation needs a deterministic identity for its unit of work — the same input must always produce the same key. This is the idempotency key: a natural key, a content hash, or a request id. Without a stable key, you cannot tell "the same work retried" from "genuinely new work."
• Sink write mode. INSERT (append) is inherently non-idempotent — two runs append two copies. upsert / MERGE keyed on the identity, delete-then-insert by partition, and overwrite-by-deterministic-path are the write modes that make the second run a no-op or an in-place update. The sink's write mode is where idempotency is won or lost.
• Replay semantics. Retry (same input, re-run after failure), replay (reprocess historical input deliberately), and backfill (fill a past window) all funnel work through the pipeline more than once. An idempotent pipeline treats all three identically: whatever the reason for the second pass, the result is unchanged.

The three places duplicates enter — the 2026 reality.

• At-least-once delivery. Every durable messaging system — Kafka, SQS, Kinesis, Pub/Sub — guarantees at-least-once delivery by default, which is a polite way of saying "you will occasionally see the same message twice." Acknowledgements can be lost after the work is done but before the ack lands; the broker redelivers. Duplicates are not a bug in the broker; they are the contract.
• Non-atomic sinks. A task that writes rows and then commits an offset (or advances a watermark) has a window between the two: crash there and the rows are written but the progress is not recorded, so the next run re-writes them. Any sink where "do the work" and "record that the work was done" are separate steps leaks duplicates on failure.
• Non-deterministic task boundaries. A batch job that selects WHERE created_at > now() - interval '1 day' reads a different set of rows every time it runs, so a retry double-counts the overlap and a backfill is impossible to reason about. Non-determinism turns "retry the same work" into "process overlapping-but-different work," which no dedup key can save.

What interviewers listen for.

• Do you distinguish "retry-safe" from "idempotent"? — retry-safe often means "it won't crash"; idempotent means "the result is unchanged." Senior signal.
• Do you name the idempotency key as the first design decision, before touching the sink? — required answer.
• Do you reach for upsert / MERGE instead of INSERT the moment re-runs are on the table? — senior signal.
• Do you state that exactly-once delivery is impossible but effectively-once processing is achievable? — required answer.
• Do you key batch work on a logical interval (execution date) rather than now()? — senior signal.

Worked example — the double-count anatomy

Detailed explanation. The single most common idempotency bug is the append-on-retry double count. A task reads a source window, appends rows to a fact table, then records progress; a crash between the append and the progress-write causes the next run to append the same rows again. Walk through the exact failure timeline so you can recognise it in a post-incident review.

• The task. Read yesterday's orders, insert them into fact_orders, then update a watermark.
• The crash point. After the INSERT commits but before the watermark advances.
• The consequence. The retry re-reads the same window and re-inserts, doubling the rows.
• The fix preview. Make the write idempotent (upsert by key) or make the boundary deterministic and overwrite the partition.

Question. Trace the state of fact_orders across an initial run that crashes mid-task and the automatic retry, for a source window containing exactly two orders.

Input.

Step	Action	Watermark after	fact_orders rows
Run 1	INSERT 2 orders	(not yet advanced)	2
Run 1	crash before watermark update	unchanged	2
Run 2 (retry)	re-read same window, INSERT 2 orders	advanced	4

Code.

# NON-IDEMPOTENT task — appends, then records progress. Crash between = duplicates.
def load_orders_bad(conn, window_start, window_end):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO fact_orders (order_id, customer_id, total_cents, order_ts)
            SELECT order_id, customer_id, total_cents, order_ts
            FROM   staging_orders
            WHERE  order_ts >= %s AND order_ts < %s
        """, (window_start, window_end))
    conn.commit()                      # <-- rows are now durable

    # ---- if the worker is OOM-killed HERE, the retry re-runs the INSERT ----

    record_watermark(conn, "fact_orders", window_end)   # progress recorded separately

Step-by-step explanation.

1. Run 1 executes the INSERT ... SELECT and commits. fact_orders now durably holds the two orders. The append itself is fine; the problem is that it is append, so it has no memory of having run.
2. The worker dies after conn.commit() but before record_watermark. The two rows are durable; the progress marker still points at the old position. From the orchestrator's view, the task failed — so it retries.
3. Run 2 reads the same window (the watermark never advanced), executes the identical INSERT ... SELECT, and commits. fact_orders now holds four rows — each order twice.
4. The watermark finally advances, so the pipeline looks healthy. The duplication is invisible until someone sums total_cents and finds revenue doubled for that window.
5. Nothing here is "wrong" in the sense of a thrown exception — every statement succeeded. The bug is structural: an append with a separate progress-write is not idempotent, and retries are guaranteed to happen.

Output.

Metric	After clean single run	After crash + retry (bug)
Row count	2	4
SUM(total_cents)	correct	2× overstated
Task status in Airflow	success	success (looks fine)
Detectability	n/a	only via row-count / revenue anomaly

Rule of thumb. Any task whose shape is "append rows, then separately record progress" is a double-count waiting for its first crash. The two safe rewrites are (a) upsert by a stable key so the re-insert is a no-op, or (b) delete-then-insert the target partition so the re-run overwrites. Never ship a bare append into a fact table that an orchestrator is allowed to retry.

Worked example — retry vs replay vs backfill

Detailed explanation. Engineers often build a pipeline that survives automatic retries but breaks the moment someone triggers a manual replay or a backfill, because they conflated the three. An idempotent pipeline handles all three with one mechanism: whatever causes the work to run again, the result is the same. Walk through the three triggers and confirm the idempotent design absorbs each.

• Retry. The orchestrator re-runs a failed task instance for the same logical interval. Cause: transient failure.
• Replay. An operator deliberately re-runs a successful interval — e.g. a bug in the transform was fixed and yesterday's output must be recomputed.
• Backfill. A range of past intervals is run for the first time, or re-run, e.g. after adding a new column that needs historical values.

Question. For a daily aggregation keyed on the execution date, show that retry, replay, and backfill all converge to the same output when the write is a partition overwrite.

Input.

Trigger	Intervals run	Expected effect
Retry	one (the failed day)	identical to a clean run of that day
Replay	one (a past, successful day)	recomputed; identical if inputs unchanged
Backfill	many past days	each day identical to a clean run of that day

Code.

# IDEMPOTENT daily aggregation — deterministic interval + partition overwrite.
def build_daily_revenue(conn, ds):          # ds = the logical date, e.g. '2026-09-05'
    with conn:                              # one transaction: delete + insert commit together
        with conn.cursor() as cur:
            # 1. Clear THIS day's partition (safe to run any number of times)
            cur.execute("DELETE FROM daily_revenue WHERE revenue_date = %s", (ds,))
            # 2. Recompute THIS day's aggregate from the immutable source window
            cur.execute("""
                INSERT INTO daily_revenue (revenue_date, customer_id, revenue_cents)
                SELECT %s, customer_id, SUM(total_cents)
                FROM   fact_orders
                WHERE  order_ts >= %s::date
                  AND  order_ts <  (%s::date + INTERVAL '1 day')
                GROUP  BY customer_id
            """, (ds, ds, ds))

Step-by-step explanation.

1. The unit of work is "one day, identified by ds." ds is the logical execution date supplied by the orchestrator, not now() — so the same day always reads the same source window and produces the same output.
2. Step 1 deletes the target partition for ds. This is the idempotency lever: deleting a possibly-nonexistent day's rows is harmless, so running it again costs nothing.
3. Step 2 recomputes the aggregate from fact_orders filtered to [ds, ds+1). The delete and the insert are wrapped in one transaction (with conn:), so a crash rolls back both — the partition is never left half-empty.
4. A retry of a failed day re-runs delete+insert for that day → identical result. A replay of a successful day does the same → identical result. A backfill loops over many days, each doing delete+insert for its own partition → each identical to a clean run.
5. There is no separate watermark to get out of sync, and no append to double-count. The partition is the progress marker: its presence and contents are fully determined by ds and the (immutable) source.

Output.

Trigger	Runs of build_daily_revenue('2026-09-05')	Final rows for 2026-09-05
Clean run	1	correct aggregate
Retry	2	same correct aggregate
Replay	3	same correct aggregate
Backfill (incl. this day)	4	same correct aggregate

Rule of thumb. Design for replay and backfill, not just automatic retry. If your task takes a logical interval as input and overwrites (or upserts) the corresponding partition, then retry, replay, and backfill are the same operation and all three are safe by construction.

Worked example — the idempotency contract checklist

Detailed explanation. Before you call a pipeline idempotent, walk a five-point checklist. Each point closes one of the duplicate-entry doors from the section intro. Treat it as the code-review rubric for any pipeline that an orchestrator or a message broker can drive more than once.

• Deterministic input boundary. The work reads a window defined by a logical parameter, not the wall clock.
• Stable identity. Each output row carries a key derived deterministically from its input.
• Idempotent write. The sink upserts by key or overwrites by partition — never bare-appends.
• Atomic progress. "Did the work" and "recorded that the work was done" commit together, or progress is implicit in the output.
• Absorbs duplicate delivery. If fed by a broker, the processor dedupes or upserts so a redelivered message is a no-op.

Question. Apply the checklist to a Kafka-to-Postgres consumer that inserts one row per message, and identify which points it fails.

Input.

Checklist point	Kafka→Postgres naive consumer
Deterministic boundary	partial (offset-driven, but commit-after-write)
Stable identity	fails (no key; relies on auto-increment PK)
Idempotent write	fails (bare INSERT)
Atomic progress	fails (write rows, then commit offset)
Absorbs duplicate delivery	fails (redelivery inserts again)

Code.

# Score a consumer against the checklist (illustrative rubric)
def idempotency_score(consumer) -> dict:
    return {
        "deterministic_boundary":   consumer.reads_by_offset_or_interval,
        "stable_identity":          consumer.has_deterministic_key,
        "idempotent_write":         consumer.write_mode in ("upsert", "overwrite"),
        "atomic_progress":          consumer.commits_progress_with_work,
        "absorbs_duplicates":       consumer.dedupes_or_upserts,
    }

naive = idempotency_score(kafka_naive_consumer)
# → all False except deterministic_boundary (partial)
print("idempotent" if all(naive.values()) else "NOT idempotent", naive)
# → NOT idempotent {...}

Step-by-step explanation.

1. The naive consumer reads by offset, which is a deterministic boundary — but it commits the offset after the write, so a crash in between replays the message. Partial credit only.
2. It has no stable identity: it relies on a database auto-increment primary key, which is generated fresh on each insert. Two inserts of "the same" message get two different PKs, so the sink cannot recognise the duplicate.
3. The write is a bare INSERT, which is the canonical non-idempotent write. This single failure is enough to sink the whole score.
4. Progress (offset commit) is separate from the work (row insert), so the crash window leaks duplicates. Atomic progress fails.
5. Because none of identity, idempotent write, or atomic progress hold, a redelivered message produces a second row. The consumer scores "NOT idempotent" and will duplicate on the first rebalance.

Output.

Checklist point	Naive	Fixed (later sections)
Deterministic boundary	partial	yes (offset window)
Stable identity	no	yes (message key / content hash)
Idempotent write	no	yes (ON CONFLICT upsert)
Atomic progress	no	yes (offset stored in same txn / dedup table)
Absorbs duplicates	no	yes (upsert is a no-op on repeat)

Rule of thumb. Run the five-point checklist on every pipeline before you trust its retries. A single "no" on idempotent write or stable identity is enough to guarantee duplicates under failure — those two are the load-bearing points, and the next two sections are entirely about getting them right.

ETL
Topic — etl
ETL problems on retry-safe pipeline design

Practice →

Data Processing Topic — data-processing Data processing problems on deduplication

Practice →

---

2. Idempotency keys & dedup stores

A deterministic idempotency key plus a store that remembers it is how you tell "the same work retried" from "genuinely new work"

The mental model in one line: an idempotency key is a deterministic identifier for a unit of work — the same input must always hash to the same key — and a dedup store is a durable seen-set keyed by that identifier, so that the first arrival of a key is admitted and every later arrival of the same key is dropped, turning an at-least-once stream into effectively-once processing at the cost of remembering which keys you have seen. Everything about deduplication reduces to two decisions: how you derive the key, and where you remember it.

Choosing the key — three sources, in priority order.

• Natural / business key. If the event already carries a unique business identifier — order_id, payment_id, (user_id, event_ts, event_type) — use it. It is the cheapest and most meaningful key because it is stable across the entire lifecycle of the record, not just this transport hop.
• Content hash. When there is no natural key, hash the content — sha256(canonical_json(payload)). Identical payloads hash identically, so a redelivered byte-for-byte copy collapses to one. The subtlety: you must canonicalise (sort keys, fix encoding) so semantically-equal payloads hash equal.
• Producer-supplied request id. For request/response systems, the client generates a UUID once per logical request and re-sends the same id on retry (an Idempotency-Key header). The server dedupes on it. The critical rule: the id is generated once per request, not once per attempt.

The cardinal sin — a fresh key per attempt.

• The bug. Generating uuid4() inside the retry path means every attempt gets a different key, so the dedup store never recognises the retry and every attempt is admitted. This is the single most common dedup failure.
• The rule. The key must be a pure function of the input, computed once and reused on every attempt. If you cannot regenerate the identical key on a retry from the same input, it is not an idempotency key.

The dedup store — where "seen" lives.

• Postgres unique index. The simplest durable seen-set: a table with a unique constraint on the key. Inserting a duplicate raises a unique-violation you catch and treat as "already seen." Transactional, so the seen-record and the work commit together.
• Redis SETNX + TTL. SET key value NX EX <ttl> returns success only for the first setter. Fast, in-memory, with a TTL that bounds how long you remember. Perfect for high-throughput streams where you only need to dedupe within a window.
• Bloom / Cuckoo filter. When the key cardinality is too large to store exactly, a probabilistic filter gives bounded memory at the cost of a small false-positive rate (it may claim "seen" for a never-seen key). Used as a first-stage filter in front of an exact store.

TTL and cardinality — the cost of memory.

• You cannot remember forever. An unbounded seen-set grows without limit. The TTL must cover the maximum possible redelivery / replay window — Kafka retention, retry horizon, backfill lookback — plus a safety margin, and no more.
• Late duplicates. If a duplicate arrives after the TTL expires, it is admitted again. The TTL is therefore a correctness parameter, not just a cost knob: size it to your redelivery guarantees.

Common interview probes on dedup.

• "Where does the idempotency key come from?" — natural key first, then content hash, then producer request id.
• "Why is uuid4() per attempt wrong?" — the retry gets a new key, so dedup never fires.
• "How do you bound the dedup store?" — TTL sized to the redelivery window; probabilistic filter for huge cardinality.
• "How do you make dedup and the work atomic?" — store the key in the same transaction as the effect (Postgres), or upsert so the write itself is the dedup.

Worked example — hashing a record into a stable key

Detailed explanation. When events lack a natural key, derive one by hashing the canonicalised content. The whole correctness of dedup rests on determinism: the same logical event, serialised twice, must produce identical bytes and therefore an identical hash. Walk through building a stable content key.

• Canonicalise. Sort object keys, use a fixed separator, encode UTF-8 — remove every source of byte-level variation.
• Hash. sha256 over the canonical bytes; hex-encode the digest as the key.
• Exclude volatile fields. Drop transport metadata (receive timestamp, delivery attempt) that differs between copies of the same event.

Question. Write a record_key function that produces the same key for two deliveries of the same logical order event, even when key ordering and transport metadata differ.

Input.

Delivery	Payload seen by consumer
First	{"order_id": 42, "amount": 1500, "_recv_ts": 1000}
Redelivery	{"amount": 1500, "order_id": 42, "_recv_ts": 1050}

Code.

import hashlib
import json

VOLATILE = {"_recv_ts", "_attempt", "_partition", "_offset"}

def record_key(payload: dict) -> str:
    """Deterministic content key: canonical JSON of business fields, sha256 hex."""
    business = {k: v for k, v in payload.items() if k not in VOLATILE}
    canonical = json.dumps(
        business,
        sort_keys=True,          # key order must not affect the hash
        separators=(",", ":"),   # fixed separators, no incidental whitespace
        ensure_ascii=False,
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()

first      = record_key({"order_id": 42, "amount": 1500, "_recv_ts": 1000})
redelivery = record_key({"amount": 1500, "order_id": 42, "_recv_ts": 1050})
print(first == redelivery)   # → True  (same logical event → same key)

Step-by-step explanation.

1. VOLATILE lists fields that describe the delivery, not the event — receive timestamp, attempt number, Kafka partition/offset. These differ between copies of the same event, so they must be excluded or the hash would differ.
2. business filters the payload down to the fields that define the event's identity. Two deliveries of the same order now share an identical business dict regardless of the transport noise around them.
3. json.dumps(..., sort_keys=True, separators=(",", ":")) canonicalises: sorting keys removes ordering variation, fixed separators remove whitespace variation, and explicit UTF-8 encoding removes charset variation. The output bytes are a pure function of the business content.
4. sha256(...).hexdigest() produces a 64-char hex string that is astronomically unlikely to collide for distinct content and is guaranteed identical for identical content.
5. The two deliveries — which differ in key order and _recv_ts — produce the same key, so the dedup store will recognise the redelivery. Had we hashed the raw payload, the differing _recv_ts alone would have produced two keys and defeated dedup.

Output.

Delivery	Business fields hashed	Key (truncated)	Dedup verdict
First	{"amount":1500,"order_id":42}	9f2c…	admit
Redelivery	{"amount":1500,"order_id":42}	9f2c…	drop (same key)

Rule of thumb. A content-hash key is only as good as its canonicalisation. Sort keys, fix separators, pin the encoding, and strip transport metadata before hashing. If you cannot reproduce the exact same key from the same logical event on a second delivery, you do not have an idempotency key — you have a random string.

Worked example — Postgres unique-index dedup gate

Detailed explanation. The most robust dedup store for transactional pipelines is a Postgres table with a unique index on the idempotency key. It is durable, transactional, and lets you commit "seen" together with the actual effect — so there is no window where the work is done but the key is not recorded. Walk through the gate.

• The table. processed_events(event_key TEXT PRIMARY KEY, processed_at TIMESTAMPTZ).
• The gate. Try to insert the key; on unique-violation, the event was already processed — skip.
• The atomicity. Do the insert of the key and the side effect in one transaction.

Question. Implement an idempotent handler that records the key and performs the side effect atomically, skipping cleanly on a duplicate.

Input.

Field	Value
Dedup table	processed_events(event_key PK, processed_at)
Effect table	fact_orders(order_id, amount_cents, ...)
Key source	record_key(payload) from the previous example
Delivery guarantee	at-least-once (duplicates possible)

Code.

import psycopg2
from psycopg2 import errors as pg_errors

def handle_event(conn, payload: dict) -> str:
    key = record_key(payload)
    try:
        with conn:                                  # BEGIN … COMMIT (ROLLBACK on error)
            with conn.cursor() as cur:
                # 1. Dedup gate — fails with UniqueViolation if key already seen
                cur.execute(
                    "INSERT INTO processed_events (event_key) VALUES (%s)", (key,)
                )
                # 2. The actual side effect — commits atomically with the gate
                cur.execute("""
                    INSERT INTO fact_orders (order_id, amount_cents, order_ts)
                    VALUES (%(order_id)s, %(amount)s, now())
                """, payload)
        return "processed"
    except pg_errors.UniqueViolation:
        conn.rollback()
        return "skipped-duplicate"

-- Dedup store: the unique index IS the gate
CREATE TABLE processed_events (
    event_key    TEXT        PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

Step-by-step explanation.

1. record_key(payload) derives the deterministic key. The same event on any delivery produces the same key, so the gate can recognise it.
2. Step 1 attempts INSERT INTO processed_events. For a first-seen key it succeeds; for a duplicate it raises UniqueViolation because the key is the primary key.
3. Step 2 performs the real side effect (writing the fact row). It runs only if step 1 succeeded, and it is inside the same transaction — so "recorded as seen" and "did the work" commit together or roll back together. There is no crash window that leaves one without the other.
4. On a duplicate, the UniqueViolation from step 1 aborts the transaction before step 2 runs, so the side effect never happens. We catch it, roll back, and report skipped-duplicate.
5. Because the gate and the effect are atomic, even a crash after commit is safe: the key is recorded, so the redelivery is skipped; a crash before commit rolls back both, so the redelivery re-processes cleanly. Every failure timeline converges to exactly one effect.

Output.

Delivery	INSERT processed_events	Side effect	Return
First	succeeds	fact row written	processed
Redelivery	UniqueViolation	not executed	skipped-duplicate
Retry after mid-txn crash	succeeds (rolled back before)	fact row written	processed

Rule of thumb. Put the dedup key and the side effect in one transaction and let a unique index be the gate. This is the gold-standard dedup pattern for anything that already writes to a transactional database — it is exactly-once processing against an at-least-once source, with no separate store to keep in sync.

Worked example — Redis SETNX with a bounded TTL

Detailed explanation. When throughput is too high for a database round-trip per event, or the sink is not transactional, use Redis as a fast dedup store. SET key val NX EX ttl sets the key only if it does not exist and expires it after ttl seconds — the first setter wins and memory is bounded by the TTL window. Walk through a high-throughput dedup gate.

• The primitive. SETNX (via SET ... NX) is atomic — exactly one concurrent setter succeeds.
• The TTL. Sized to the maximum redelivery window (e.g. Kafka retention + retry horizon).
• The trade-off. Not transactional with the sink, so pair it with an idempotent sink write for end-to-end safety.

Question. Implement a Redis dedup gate with a TTL, and explain why the TTL is a correctness parameter, not just a memory knob.

Input.

Parameter	Value
Store	Redis
Primitive	SET key 1 NX EX 604800 (7-day TTL)
Redelivery window	≤ 7 days (Kafka retention)
Sink	idempotent (upsert) as backstop

Code.

import redis

r = redis.Redis(host="redis", port=6379)
DEDUP_TTL_SECONDS = 7 * 24 * 3600      # must cover the max redelivery window

def seen_before(key: str) -> bool:
    """Return True if this key was already processed within the TTL window."""
    # SET NX returns True only for the first setter; None if the key already exists.
    was_set = r.set(name=f"dedup:{key}", value=1, nx=True, ex=DEDUP_TTL_SECONDS)
    return was_set is None

def process(payload: dict) -> str:
    key = record_key(payload)
    if seen_before(key):
        return "skipped-duplicate"
    do_side_effect(payload)            # itself idempotent (upsert) as a backstop
    return "processed"

Step-by-step explanation.

1. r.set(..., nx=True, ex=TTL) attempts to create the key only if absent. Redis executes this atomically, so under concurrent duplicate deliveries exactly one caller gets True (created) and the rest get None (already exists).
2. seen_before returns True when the key already existed — meaning this is a duplicate within the TTL window — and False for the first arrival.
3. process derives the deterministic key, checks the gate, and only runs the side effect for a first arrival. Duplicates short-circuit to skipped-duplicate.
4. The TTL is a correctness parameter: if a duplicate can be redelivered up to 7 days later (Kafka retention), the TTL must be ≥ 7 days, or a late duplicate outlives its key and is admitted again. Sizing the TTL below the redelivery window silently reintroduces duplicates.
5. Redis is not transactional with the sink, so there is a small window: gate passes, worker crashes before the side effect, key remains set → the event is lost, not duplicated. To cover that, the side effect is itself idempotent (an upsert), so even if the gate is bypassed the sink absorbs the repeat. Belt and braces.

Output.

Scenario	Gate result	Side effect	Net
First arrival	created (False = not seen)	runs	one effect
Duplicate within TTL	exists (True = seen)	skipped	no extra effect
Duplicate after TTL	created again	runs — but upsert sink absorbs it	still one effect
Crash after gate, before effect	key set	lost by gate, recovered by re-drive + upsert	one effect

Rule of thumb. Use Redis SET NX EX for high-throughput dedup, size the TTL to the maximum redelivery/replay window, and always back it with an idempotent sink. A fast dedup store handles the common case cheaply; the idempotent sink is the correctness guarantee that survives the store's non-transactional edge cases.

Data engineering interview question on idempotency keys and dedup

A senior interviewer might ask: "You consume an at-least-once Kafka topic of payment events and must apply each payment to a Postgres ledger exactly once, even across consumer restarts and partition rebalances. There is no natural request id on the events. Design the idempotency key, the dedup store, and the write so that no payment is ever double-applied or lost, and bound the memory the dedup uses."

Solution Using a content-hash key + transactional dedup gate + idempotent ledger upsert

import hashlib, json
import psycopg2
from psycopg2 import errors as pg_errors

VOLATILE = {"_recv_ts", "_attempt", "_partition", "_offset"}

def payment_key(evt: dict) -> str:
    business = {k: v for k, v in evt.items() if k not in VOLATILE}
    canonical = json.dumps(business, sort_keys=True, separators=(",", ":")).encode()
    return hashlib.sha256(canonical).hexdigest()

def apply_payment(conn, evt: dict) -> str:
    key = payment_key(evt)
    try:
        with conn:                                    # single transaction
            with conn.cursor() as cur:
                # 1. Dedup gate (unique index on event_key)
                cur.execute(
                    "INSERT INTO processed_payments (event_key) VALUES (%s)", (key,)
                )
                # 2. Apply to the ledger — upsert so it is a no-op if somehow re-run
                cur.execute("""
                    INSERT INTO ledger (payment_id, account_id, amount_cents)
                    VALUES (%(payment_id)s, %(account_id)s, %(amount_cents)s)
                    ON CONFLICT (payment_id) DO NOTHING
                """, evt)
        return "applied"
    except pg_errors.UniqueViolation:
        conn.rollback()
        return "skipped-duplicate"

-- Dedup store + ledger, both keyed for idempotency
CREATE TABLE processed_payments (
    event_key    TEXT        PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

CREATE TABLE ledger (
    payment_id   BIGINT      PRIMARY KEY,      -- natural key inside the payload
    account_id   BIGINT      NOT NULL,
    amount_cents BIGINT      NOT NULL,
    applied_at   TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

-- Bound dedup memory: TTL the gate table to the redelivery window (e.g. 30 days)
DELETE FROM processed_payments WHERE processed_at < now() - INTERVAL '30 days';

Step-by-step trace.

Step	Input	Reasoning
Derive key	sha256(canonical business fields)	deterministic; same event → same key across restarts
Gate insert	INSERT processed_payments	unique index rejects a repeat within the txn
Ledger write	ON CONFLICT (payment_id) DO NOTHING	second layer: even a gap in the gate cannot double-apply
Commit	gate + ledger together	atomic: seen ⇔ applied, no crash window
Duplicate	UniqueViolation on gate	roll back, skip cleanly
Memory bound	nightly DELETE older than 30 days	seen-set stays O(30 days of events)

After deployment, each payment is applied exactly once. A consumer restart replays uncommitted offsets → the gate recognises already-applied keys and skips; a partition rebalance redelivers in-flight messages → the ledger's ON CONFLICT DO NOTHING makes the repeat a no-op even in the rare window where the gate has not yet caught up. Two independent idempotency layers (content-hash gate + natural-key ledger PK) mean no single failure double-applies money.

Output:

Metric	Value
Applied per unique payment	exactly 1
Double-apply on rebalance	0 (gate + upsert)
Lost payment on restart	0 (offsets replay; gate/upsert absorb repeats)
Dedup memory	O(events in last 30 days)
Extra write cost	1 gate row + 1 upsert per payment

Why this works — concept by concept:

• Content-hash idempotency key — with no request id, sha256 over canonicalised business fields gives a stable identity that survives restarts and rebalances. The same payment always hashes to the same key, so "seen" is well-defined.
• Transactional dedup gate — the unique index on processed_payments.event_key makes "record seen" and "apply payment" commit atomically. There is no window where the payment is applied but not recorded, so a crash can never leave the two out of sync.
• Natural-key upsert backstop — ledger.payment_id is the primary key and the ledger write is ON CONFLICT DO NOTHING. Even if the gate were bypassed, the ledger physically cannot hold two rows for one payment. Two layers, independent failure modes.
• TTL on the gate table — the seen-set is bounded to the redelivery window (30 days ≫ Kafka retention). Beyond that, late duplicates are absorbed by the ledger PK, so trimming the gate is safe and memory stays flat.
• Cost — one extra gate row and one upsert per payment: O(1) per event, O(N) storage bounded by the TTL window. Compared to the correctness disaster of double-applied payments, the overhead is negligible. The design is exactly-once processing over an at-least-once source — the only kind of exactly-once that actually exists.

Data Processing
Topic — data-processing
Data processing problems on deduplication and keys

Practice →

Streaming Topic — streaming Streaming problems on at-least-once dedup

Practice →

---

3. Upsert / MERGE & idempotent sinks

INSERT is not idempotent; upsert keyed on the identity is — the primary key is the idempotency contract at the sink

The mental model in one line: a bare INSERT appends a new row on every run and is therefore never idempotent, while an upsert (INSERT ... ON CONFLICT DO UPDATE, MERGE, or delete-then-insert by partition) is idempotent because it keys on the row's identity and either updates the existing row in place or does nothing — so re-running the load converges to one row per key no matter how many times it fires. The single most impactful change you can make to a pipeline's retry safety is at the sink: stop appending, start upserting.

Why insert fails and upsert wins.

• Insert has no memory. INSERT INTO fact SELECT ... produces new rows every time; the database has no way to know these rows "are the same" as a previous run's. Two runs → two copies. This is non-idempotency by construction.
• Upsert keys on identity. INSERT ... ON CONFLICT (id) DO UPDATE requires a unique key; on the second run the conflict fires and the row is updated in place (or ignored) instead of duplicated. The number of rows is a function of the distinct keys, not the number of runs.
• The key IS the contract. Upsert idempotency is only as strong as the uniqueness of the conflict target. If two logically-distinct rows can share the same key, upsert merges them incorrectly; if two copies of the same row get different keys, upsert duplicates them. Get the key right.

The three idempotent-write strategies.

• MERGE / ON CONFLICT by key. Best when the unit of work is a set of rows identified by natural keys and you want in-place updates (SCD, late-arriving corrections). One statement, per-key semantics.
• Delete-insert by partition. Best when the unit of work is a whole partition (a day, an hour, a batch). Delete the partition, insert the fresh computation, in one transaction. Simpler than MERGE, and naturally handles rows that disappeared from the source (MERGE alone leaves stale rows behind).
• Overwrite by deterministic path. Best for object storage / lakes: write the batch to a path derived from the logical interval (.../date=2026-09-05/) and overwrite it. The path is the key; re-running overwrites the same location.

Choosing between MERGE and delete-insert.

• MERGE handles inserts, updates, and (optionally) deletes per key, but a naive MERGE does not remove target rows whose key vanished from the source — you need an explicit WHEN NOT MATCHED BY SOURCE THEN DELETE for full-refresh semantics.
• Delete-insert is a full replace of the partition: any row not in the new computation is gone, which is exactly right for "recompute this day from scratch." It is the simpler mental model when the whole partition is the unit.

Idempotent producers and side effects.

• Kafka idempotent producer. enable.idempotence=true gives each producer a PID + sequence number so the broker dedupes retried produce requests — a single message is written once even if the produce is retried.
• Non-DB side effects. Sending an email, charging a card, calling an external API — wrap these behind an idempotency key the downstream honours (Idempotency-Key header), or gate them with a dedup store so the second attempt is suppressed.

Common interview probes on idempotent sinks.

• "Why isn't INSERT idempotent?" — no memory of prior runs; appends duplicates.
• "Upsert vs delete-insert?" — upsert for per-key updates; delete-insert for full-partition recompute and vanished-row handling.
• "What if the source row disappears?" — MERGE needs WHEN NOT MATCHED BY SOURCE DELETE; delete-insert handles it for free.
• "How do you make writing to S3 idempotent?" — deterministic path per interval + overwrite (or atomic rename of a staged file).

Worked example — Postgres ON CONFLICT upsert

Detailed explanation. The canonical idempotent sink write in Postgres is INSERT ... ON CONFLICT (key) DO UPDATE. It requires a unique index on the conflict target; on a repeat, the existing row is updated instead of a duplicate being inserted. Walk through upserting a batch of customer dimension rows.

• The target. dim_customer(customer_id PK, name, tier, updated_at).
• The write. Upsert on customer_id; on conflict, update the mutable columns.
• The guarantee. Re-running the identical batch leaves the table byte-for-byte identical.

Question. Write an upsert that loads a customer batch idempotently, and show the table state after running it twice.

Input.

customer_id	name	tier
1	Ada	gold
2	Grace	silver

Code.

-- Target with a unique key = the idempotency contract
CREATE TABLE dim_customer (
    customer_id BIGINT      PRIMARY KEY,
    name        TEXT        NOT NULL,
    tier        TEXT        NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

-- Idempotent load: upsert by customer_id
INSERT INTO dim_customer (customer_id, name, tier)
VALUES (1, 'Ada', 'gold'),
       (2, 'Grace', 'silver')
ON CONFLICT (customer_id) DO UPDATE
    SET name       = EXCLUDED.name,
        tier       = EXCLUDED.tier,
        updated_at = clock_timestamp();

Step-by-step explanation.

1. dim_customer has customer_id as its primary key — the unique constraint that ON CONFLICT targets. Without a unique key there is no conflict to detect and the clause is meaningless.
2. On the first run, neither key exists, so both rows are inserted. The table holds two rows.
3. On the second run, both customer_ids already exist, so ON CONFLICT (customer_id) DO UPDATE fires for each. Instead of inserting duplicates, Postgres updates the existing rows using EXCLUDED (the would-be-inserted values).
4. Because the batch values are identical, the update writes the same name and tier back; only updated_at changes. The row count is unchanged — still two rows, one per key.
5. This makes the load idempotent on the dimension: run it once, twice, or a hundred times and you always end with exactly one row per customer_id, carrying the latest batch's values.

Output.

Run	Rows in dim_customer	Behaviour
After run 1	2	both inserted
After run 2	2	both updated in place
After run N	2	idempotent — one row per key

Rule of thumb. Every dimension and keyed-fact load should be an ON CONFLICT DO UPDATE upsert on a real unique key. If the batch can also delete keys, add a reconciliation step; if it only ever adds/updates, the upsert alone is fully idempotent.

Worked example — partition overwrite (delete-insert)

Detailed explanation. When the unit of work is a whole partition — one day of a fact table — the cleanest idempotent write is delete-then-insert inside one transaction. It handles new rows, changed rows, and rows that disappeared from the source, because the entire partition is replaced. Walk through a daily fact rebuild.

• The unit. One day, identified by the logical date ds.
• The write. DELETE WHERE date = ds; INSERT SELECT ... WHERE date = ds; in one transaction.
• The advantage over MERGE. Vanished source rows are removed for free — the delete clears them.

Question. Write a delete-insert that rebuilds one day of fact_sales idempotently, and trace a re-run where one source row was deleted between runs.

Input.

Run	Source rows for 2026-09-05
1	orders 100, 101, 102
2 (order 101 refunded/removed)	orders 100, 102

Code.

-- Idempotent daily rebuild — delete the day, then reinsert it, atomically.
BEGIN;

DELETE FROM fact_sales
WHERE  sale_date = DATE '2026-09-05';

INSERT INTO fact_sales (sale_date, order_id, amount_cents)
SELECT DATE '2026-09-05', order_id, amount_cents
FROM   staging_orders
WHERE  order_ts >= DATE '2026-09-05'
  AND  order_ts <  DATE '2026-09-05' + INTERVAL '1 day';

COMMIT;

Step-by-step explanation.

1. The BEGIN ... COMMIT wraps both statements so they are atomic: a crash between the delete and the insert rolls back the delete, leaving the previous partition intact. The partition is never left empty.
2. DELETE ... WHERE sale_date = '2026-09-05' clears the day's partition. On the first run it deletes nothing (empty); on a re-run it clears the prior computation. Deleting a non-existent day is harmless — that is the idempotency lever.
3. INSERT ... SELECT recomputes the day from the immutable source window [2026-09-05, 2026-09-06). The output is a pure function of the source and the date.
4. On run 2, order 101 has been removed from the source. The delete clears the old three rows; the insert writes only the two surviving rows. The partition now correctly holds 100 and 102 — the vanished row is gone with no special handling.
5. Run the whole thing any number of times and the partition always equals "the current source, aggregated for that day." Retry, replay, and backfill are all the same delete-insert.

Output.

After run	Rows for 2026-09-05	Note
Run 1	100, 101, 102	initial build
Run 2	100, 102	101 removed for free by delete
Any re-run	matches current source	idempotent full-partition replace

Rule of thumb. For fact tables partitioned by time, prefer delete-insert per partition inside one transaction. It is simpler than MERGE, it handles disappearing rows automatically, and the partition itself becomes the idempotency unit — no separate watermark or dedup store required.

Worked example — Snowflake / warehouse MERGE by key

Detailed explanation. In cloud warehouses the idiomatic idempotent sink is MERGE, which expresses insert-or-update (and optionally delete) in one statement. The critical detail is deduping the source before the merge: MERGE errors (or non-deterministically picks) when multiple source rows match one target key. Walk through a merge that loads a CDC-style delta idempotently.

• The source. A staged delta that may itself contain duplicate keys (at-least-once upstream).
• The dedup. QUALIFY ROW_NUMBER() ... = 1 to keep the latest row per key before merging.
• The merge. Update matched keys, insert new keys, in one statement.

Question. Write a MERGE that upserts a deduplicated delta into a target dimension, keeping the latest version per key.

Input.

order_id	status	event_ts
42	shipped	10:00
42	delivered	10:05
43	shipped	10:02

Code.

-- Warehouse MERGE — dedupe the source, then upsert by key.
MERGE INTO dim_order AS tgt
USING (
    SELECT order_id, status, event_ts
    FROM   staging_order_delta
    QUALIFY ROW_NUMBER() OVER (
                PARTITION BY order_id
                ORDER BY event_ts DESC        -- keep the latest per key
            ) = 1
) AS src
ON tgt.order_id = src.order_id
WHEN MATCHED THEN UPDATE SET
    tgt.status   = src.status,
    tgt.event_ts = src.event_ts
WHEN NOT MATCHED THEN INSERT (order_id, status, event_ts)
    VALUES (src.order_id, src.status, src.event_ts);

Step-by-step explanation.

1. The staged delta contains two rows for order_id = 42 (an at-least-once upstream, or two status changes in the batch). Feeding both to MERGE would be ambiguous — which one wins?
2. QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY event_ts DESC) = 1 collapses the source to one row per key, keeping the latest by event_ts. Now the merge source has a unique key.
3. ON tgt.order_id = src.order_id matches by the idempotency key. WHEN MATCHED updates the existing dimension row; WHEN NOT MATCHED inserts a new one.
4. Because the source is deduped and the match is by key, running the MERGE twice is idempotent: the second run matches the same keys and writes the same values, leaving row counts unchanged.
5. If you also need to remove keys deleted upstream, add WHEN NOT MATCHED BY SOURCE THEN DELETE — but only when the source is a full snapshot, not an incremental delta (a delta legitimately omits unchanged keys).

Output.

order_id	status	Source rows collapsed
42	delivered	2 → 1 (latest kept)
43	shipped	1

Rule of thumb. Always dedupe the MERGE source to one row per key (ROW_NUMBER = 1 on the latest) before merging. A MERGE fed a source with duplicate keys is either an error or a coin flip; a deduped source makes MERGE a clean, idempotent, per-key upsert.

Data engineering interview question on idempotent sinks

A senior interviewer might ask: "Your nightly job loads a partitioned fact table from a staging area that is populated by an at-least-once ingestion, so staging can contain duplicate rows and occasionally re-processes a past day. Design the warehouse load so that re-running any night — or backfilling a range of nights — never produces duplicates and correctly reflects rows that were deleted upstream. Cover the write strategy, the source dedup, and how you'd verify idempotency."

Solution Using per-partition delete-insert with source dedup and a verification query

-- 1. Idempotent load procedure for one logical day (:ds)
--    Deletes the day's partition, then inserts a deduplicated recomputation.
BEGIN;

DELETE FROM fact_orders
WHERE  order_date = :ds;

INSERT INTO fact_orders (order_date, order_id, customer_id, amount_cents)
SELECT :ds AS order_date,
       order_id,
       customer_id,
       amount_cents
FROM (
    SELECT order_id, customer_id, amount_cents, order_ts,
           ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_ts DESC) AS rn
    FROM   staging_orders
    WHERE  order_ts >= :ds::date
      AND  order_ts <  :ds::date + INTERVAL '1 day'
) deduped
WHERE rn = 1;                       -- one row per order_id, latest wins

COMMIT;

-- 2. Verification: idempotency means zero duplicate keys per partition
SELECT order_date, order_id, COUNT(*) AS copies
FROM   fact_orders
WHERE  order_date = :ds
GROUP  BY order_date, order_id
HAVING COUNT(*) > 1;               -- MUST return zero rows

# 3. Backfill driver — loops days, each an independent idempotent load
from datetime import date, timedelta

def backfill(run_load, start: date, end: date):
    d = start
    while d <= end:
        run_load(ds=d.isoformat())     # delete-insert for day d; safe to re-run
        d += timedelta(days=1)

Step-by-step trace.

Step	Input	Reasoning
Boundary	:ds logical date	deterministic window [ds, ds+1); not now()
Source dedup	ROW_NUMBER() = 1 on latest	collapses at-least-once staging duplicates
Delete partition	DELETE WHERE order_date = :ds	clears prior run; removes vanished rows
Insert deduped	INSERT SELECT ... rn = 1	one row per order_id for the day
Atomicity	BEGIN … COMMIT	crash rolls back both; partition never half-written
Verify	HAVING COUNT(*) > 1	asserts zero duplicate keys post-load

After deployment, any single night can be re-run and any range can be backfilled with the same procedure. Staging duplicates are collapsed by ROW_NUMBER; the per-partition delete removes both the prior run's rows and any upstream-deleted orders; the transaction guarantees the partition is never left in a partial state. The verification query is wired into the DAG as a data-quality check that fails the run if a single duplicate key slips through.

Output:

Metric	Value
Duplicate keys per partition after any re-run	0
Upstream-deleted rows after re-run	removed
Backfill of N days	N independent idempotent loads
Staging duplicates in output	collapsed to latest per key
Partial-partition state on crash	impossible (transactional)

Why this works — concept by concept:

• Deterministic partition boundary — keying the load on :ds (a logical date) instead of now() means the same day always reads the same source window, so re-runs and backfills are reproducible rather than reading a moving target.
• Source dedup with ROW_NUMBER — the at-least-once staging area can hold duplicate keys; ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_ts DESC) = 1 collapses them to the latest, so the insert writes exactly one row per key.
• Delete-insert per partition — replacing the whole partition handles inserts, updates, and deletions in one move; a row that vanished upstream is simply absent from the reinsert, with no special-case logic.
• Transactional atomicity — wrapping delete + insert in one transaction means a failure rolls back to the last good partition; there is no crash window that leaves the day empty or doubled.
• Cost — O(rows in the day) per run for the delete and the insert, plus one dedup sort. Compared to a MERGE it is simpler and vanished-row-safe; compared to a bare append it trades a cheap per-partition delete for total re-run safety. The verification query is O(rows in the day) and turns "we think it's idempotent" into "the DAG proves it every night."

ETL
Topic — etl
ETL problems on upsert and MERGE loads

Practice →

Data Processing Topic — data-processing Data processing problems on partition overwrite

Practice →

---

4. Exactly-once vs at-least-once + checkpoints

exactly-once is not a delivery guarantee — it is at-least-once delivery plus idempotent processing plus checkpointing, which together produce effectively-once

The mental model in one line: no distributed system can guarantee a message is delivered exactly once — networks lose acknowledgements, so a sender must choose between at-most-once (risk loss) and at-least-once (risk duplicates) — but you can achieve effectively-once processing by combining at-least-once delivery with idempotent processing (dedup or upsert) and checkpointed progress that commits atomically with the output, so that duplicate deliveries produce exactly one effect. "Exactly-once" as marketed by streaming engines is this composition, not a delivery miracle.

The three delivery guarantees.

• At-most-once. Fire and forget: the sender does not retry, so a lost message is simply lost. Zero duplicates, possible data loss. Acceptable only for lossy metrics.
• At-least-once. The sender retries until acknowledged, so a lost ack causes redelivery. Zero loss, possible duplicates. This is the default of every durable broker and the foundation you build exactly-once on top of.
• Exactly-once (effectively-once). At-least-once delivery + idempotent processing. The duplicates from redelivery are absorbed by dedup/upsert; the loss is prevented by retry. The net effect is one-per-event even though the wire saw duplicates.

The end-to-end contract — three things must line up.

• Source position (offsets). The consumer must be able to resume from a known position (Kafka offset, Kinesis sequence number) after a crash, so no input is skipped.
• Processing state (checkpoint). Any accumulated state (running aggregates, windows) must be checkpointed so it can be restored on restart without recomputation or loss.
• Sink commit (atomic with progress). The output write and the advance of the source position must commit together — either both or neither — or the crash window between them leaks duplicates (write done, offset not committed → reprocess) or loss (offset committed, write not done → skip).

Why "exactly-once delivery" is a myth.

• The two-generals problem. A sender can never know its message was received without an ack, and the ack itself can be lost. So it must either not retry (at-most-once) or retry (at-least-once). There is no third option on the wire.
• The resolution. Move the guarantee from delivery to effect. Make processing idempotent so a redelivered message changes nothing, and the observable behaviour is exactly-once even though delivery was at-least-once.

How engines implement effectively-once.

• Kafka transactions (read-process-write). Consume, process, and produce (plus commit offsets) inside one Kafka transaction with isolation.level=read_committed downstream. Either the whole cycle commits or it aborts; consumers never see aborted output.
• Spark Structured Streaming. A write-ahead-log checkpoint records offsets and state per micro-batch; combined with an idempotent sink (foreachBatch doing a MERGE, or a deterministic file output), a re-run of a batch produces the same output.
• Flink two-phase commit. Barriers align a distributed snapshot of all operator state; the sink pre-commits on snapshot and finalises on the checkpoint-complete notification, giving transactional exactly-once into supporting sinks.

Common interview probes on exactly-once.

• "Is exactly-once delivery real?" — no; it's at-least-once delivery + idempotent processing = effectively-once.
• "What three things must be atomic?" — output write, state checkpoint, and source-offset advance.
• "Where do duplicates come from even with checkpointing?" — the window between writing output and committing the offset, unless they commit together.
• "Kafka exactly-once — how?" — transactional producer + consumer offsets in the transaction + read_committed consumers.

Worked example — at-least-once + dedup = effectively-once

Detailed explanation. The cleanest demonstration of the exactly-once equation is a consumer over an at-least-once source that dedupes by key before applying an effect. The delivery layer is allowed to duplicate; the processing layer absorbs it. Walk through the composition explicitly.

• Delivery. At-least-once: the same offset may be delivered twice after a crash.
• Processing. Dedup by idempotency key (from section 2) before the effect.
• Net. One effect per unique event — effectively-once.

Question. Show, message by message, that a duplicated at-least-once delivery yields exactly one effect when the processor dedupes by key.

Input.

Wire delivery	Event key	Prior seen?
1	A	no
2	B	no
3 (redelivery)	A	yes
4	C	no

Code.

def effectively_once(stream, seen: set, apply_effect) -> int:
    """At-least-once stream + dedup set = effectively-once effects."""
    effects = 0
    for msg in stream:                      # at-least-once: may repeat a key
        key = record_key(msg.payload)       # deterministic key (section 2)
        if key in seen:
            continue                        # duplicate delivery → no effect
        apply_effect(msg)                   # the single, real effect
        seen.add(key)                       # remember it (durable store in prod)
        effects += 1
    return effects

Step-by-step explanation.

1. Delivery 1 (key A) is unseen → apply the effect, record A. One effect so far.
2. Delivery 2 (key B) is unseen → apply, record B. Two effects.
3. Delivery 3 is a redelivery of key A (the broker's at-least-once behaviour after a lost ack). A is already in seen → skip. No new effect. This is where duplication is absorbed.
4. Delivery 4 (key C) is unseen → apply, record C. Three effects for three distinct events, despite four wire deliveries.
5. The delivery layer did its at-least-once job (four deliveries, one a duplicate); the processing layer did its idempotent job (three effects). Composed, the observable behaviour is exactly-once. In production seen is a durable store (section 2) so it survives restarts.

Output.

Distinct events	Wire deliveries	Effects applied
3 (A, B, C)	4	3

Rule of thumb. Stop chasing exactly-once delivery. Accept at-least-once delivery, put a durable dedup (or an upsert sink) in the processing path, and you get effectively-once — the only exactly-once that is physically achievable across a network.

Worked example — Spark Structured Streaming checkpoint + foreachBatch MERGE

Detailed explanation. Spark Structured Streaming gives effectively-once by pairing its checkpoint (offsets + state in a write-ahead log) with an idempotent sink. The idiomatic sink for a keyed upsert is foreachBatch running a MERGE, with the batch's determinism ensuring a replayed batch produces the same output. Walk through the streaming upsert.

• Checkpoint. checkpointLocation stores the offsets and state per micro-batch.
• Sink. foreachBatch runs a Delta/warehouse MERGE keyed on the natural key.
• Replay safety. A failed batch is replayed with the same offsets → the MERGE upserts the same keys → idempotent.

Question. Configure a Structured Streaming query that reads Kafka and upserts into a keyed Delta table idempotently.

Input.

Component	Value
Source	Kafka topic orders (at-least-once)
Checkpoint	s3://ckpt/orders/
Sink	Delta db.dim_order, key order_id
Write	foreachBatch MERGE

Code.

from pyspark.sql import functions as F

def upsert_batch(batch_df, batch_id):
    # Dedupe within the micro-batch, then MERGE by key (idempotent per batch)
    latest = (batch_df
              .withColumn("rn", F.expr(
                  "row_number() over (partition by order_id order by event_ts desc)"))
              .filter("rn = 1").drop("rn"))
    latest.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
        MERGE INTO db.dim_order AS t
        USING updates AS s
        ON  t.order_id = s.order_id
        WHEN MATCHED     THEN UPDATE SET t.status = s.status, t.event_ts = s.event_ts
        WHEN NOT MATCHED THEN INSERT (order_id, status, event_ts)
                              VALUES (s.order_id, s.status, s.event_ts)
    """)

(spark.readStream
      .format("kafka")
      .option("subscribe", "orders")
      .load()
      .selectExpr("CAST(value AS STRING) AS json")
      # ... parse json into order_id, status, event_ts ...
      .writeStream
      .foreachBatch(upsert_batch)
      .option("checkpointLocation", "s3://ckpt/orders/")   # offsets + state
      .start())

Step-by-step explanation.

1. readStream ... format("kafka") consumes at-least-once; Spark tracks the consumed offsets in the checkpoint directory, not in Kafka's own consumer group, so restarts resume precisely.
2. Each micro-batch invokes upsert_batch(df, batch_id). Structured Streaming guarantees that if a batch fails, it is retried with the identical input (same offsets) — batch_id is stable across retries.
3. upsert_batch first dedupes within the batch (row_number = 1 on latest per order_id) so the MERGE source has unique keys, then runs the MERGE keyed on order_id.
4. Because the MERGE is keyed and the retried batch carries the same rows, re-running a failed batch upserts the same keys to the same values — the output is unchanged. Idempotent sink + deterministic batch input = effectively-once.
5. The checkpoint is the linchpin: it records "batch N covered offsets X–Y" atomically with batch completion. On restart, Spark knows exactly which batch to resume, and the MERGE absorbs any partial re-application. There is no separate offset commit to fall out of sync.

Output.

Event	Delivery	Effect on dim_order
order 42 shipped	once	1 row (insert)
order 42 delivered	once	same row (update)
batch replay after failure	same offsets	same MERGE → no change

Rule of thumb. In Spark Structured Streaming, effectively-once = a stable checkpointLocation + an idempotent foreachBatch (MERGE by key or deterministic overwrite). The checkpoint handles offsets and state; the idempotent sink handles the replayed-batch case. Never write a streaming append into a keyed table.

Worked example — Kafka read-process-write transaction

Detailed explanation. When both the source and sink are Kafka, the exactly-once story is a transactional producer that writes output records and the consumer offsets inside one transaction, with downstream consumers reading read_committed. A crash aborts the transaction, so partial output is never visible. Walk through the transactional cycle.

• Producer. enable.idempotence=true, transactional.id=<stable>; wraps output + offsets in beginTransaction/commitTransaction.
• Offsets in the txn. sendOffsetsToTransaction ties consumed offsets to the same commit.
• Downstream. isolation.level=read_committed so aborted batches are invisible.

Question. Sketch a Kafka read-process-write loop that is effectively-once end to end.

Input.

Component	Value
Input topic	orders
Output topic	orders_enriched
transactional.id	enricher-1
Downstream isolation	read_committed

Code.

from confluent_kafka import Producer, Consumer

producer = Producer({
    "bootstrap.servers": "kafka:9092",
    "enable.idempotence": True,        # dedupe retried produce requests
    "transactional.id": "enricher-1",  # stable ID → fencing across restarts
})
producer.init_transactions()

consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "enricher",
    "enable.auto.commit": False,       # offsets committed inside the txn instead
    "isolation.level": "read_committed",
})
consumer.subscribe(["orders"])

while True:
    msgs = consumer.consume(num_messages=500, timeout=1.0)
    if not msgs:
        continue
    producer.begin_transaction()
    try:
        for m in msgs:
            producer.produce("orders_enriched", value=enrich(m.value()))
        # Commit consumed offsets AS PART OF the same transaction
        producer.send_offsets_to_transaction(
            consumer.position(consumer.assignment()),
            consumer.consumer_group_metadata())
        producer.commit_transaction()          # output + offsets: all or nothing
    except Exception:
        producer.abort_transaction()           # partial output never visible
        raise

Step-by-step explanation.

1. enable.idempotence=true gives the producer a PID + per-partition sequence numbers, so a retried produce (after a lost ack) is deduped by the broker — a single logical message is written once.
2. transactional.id="enricher-1" is stable across restarts; Kafka uses it to fence a zombie previous instance, so a restarted enricher cannot have its in-flight transaction committed by a stale process.
3. Each poll batch runs inside begin_transaction() … commit_transaction(). The enriched outputs are produced, and the consumed offsets are added to the transaction via send_offsets_to_transaction.
4. commit_transaction() commits the output records and the offsets atomically. If the process crashes before commit, the transaction is aborted: neither the outputs nor the offset advance are visible, so the batch is cleanly reprocessed.
5. Downstream consumers set isolation.level=read_committed, so they never read records from an aborted transaction. End to end, each input contributes exactly one output — effectively-once across the Kafka pipeline.

Output.

Scenario	Output records	Offset advance
Clean batch	committed once	committed with output
Crash before commit	aborted (invisible)	not advanced → reprocess
Producer retry (lost ack)	deduped by idempotent producer	n/a

Rule of thumb. For Kafka-to-Kafka, use a transactional producer with a stable transactional.id, put consumed offsets inside the transaction, and read read_committed downstream. That is the only way to make the output write and the offset advance atomic — which is the whole game for effectively-once.

Streaming interview question on exactly-once

A senior interviewer might ask: "You run a streaming consumer that maintains a per-user running total and writes it to a serving store. On a consumer restart, finance reports that some users' totals jumped. Explain where the double-counting comes from, then design the consumer so that a restart mid-batch cannot double-count — cover the delivery guarantee, the checkpoint, and the sink write, and be explicit about what must be atomic."

Solution Using checkpointed offsets committed atomically with an idempotent upsert sink

# Effectively-once running total: dedup by key + checkpoint offset WITH the write.
import psycopg2

def consume_loop(consumer, conn):
    while True:
        batch = consumer.poll_batch(max_records=1000)   # at-least-once source
        if not batch:
            continue
        with conn:                                       # ONE transaction
            with conn.cursor() as cur:
                for msg in batch:
                    key = record_key(msg.value)          # deterministic identity
                    # 1. Dedup gate — skip if this delta was already applied
                    cur.execute(
                        "INSERT INTO applied_deltas (event_key) VALUES (%s) "
                        "ON CONFLICT DO NOTHING", (key,))
                    if cur.rowcount == 0:
                        continue                         # duplicate → no effect
                    # 2. Apply the delta to the running total (idempotent via gate)
                    cur.execute("""
                        INSERT INTO user_totals (user_id, total_cents)
                        VALUES (%(user_id)s, %(amount_cents)s)
                        ON CONFLICT (user_id) DO UPDATE
                            SET total_cents = user_totals.total_cents + EXCLUDED.total_cents
                    """, msg.value)
                # 3. Commit the source offset IN THE SAME TRANSACTION as the writes
                cur.execute("""
                    INSERT INTO stream_offsets (partition, committed_offset)
                    VALUES (%s, %s)
                    ON CONFLICT (partition) DO UPDATE SET committed_offset = EXCLUDED.committed_offset
                """, (batch.partition, batch.end_offset))
        # transaction committed → offsets + effects are durable together

-- Progress and dedup both live in the sink DB → atomic with the effect
CREATE TABLE applied_deltas (event_key TEXT PRIMARY KEY,
                             applied_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp());
CREATE TABLE stream_offsets (partition INT PRIMARY KEY, committed_offset BIGINT NOT NULL);
CREATE TABLE user_totals   (user_id BIGINT PRIMARY KEY, total_cents BIGINT NOT NULL);

Step-by-step trace.

Step	Action	Reasoning
Poll	at-least-once batch from partition p	duplicates and replays possible
Dedup gate	INSERT applied_deltas ON CONFLICT DO NOTHING	additive delta must apply once, so gate is required
Apply delta	upsert total += amount	running total; gate makes the += safe
Commit offset	write stream_offsets in same txn	offset advance atomic with the effect
Crash before commit	whole txn rolls back	offset not advanced → batch reprocessed, gate skips applied deltas
Restart	resume from stream_offsets	no skipped input, no double-applied delta

The double-counting in the original design came from committing the Kafka offset separately from (and after) the write: a crash in that window replayed the batch and re-added the deltas. Here the offset lives in the same Postgres transaction as the running-total upsert and the dedup gate, so the three advance atomically. Note the subtlety: the running total is an additive effect, so upsert-by-key alone is not idempotent for it — the applied_deltas gate is what makes each delta apply exactly once.

Output:

Scenario	user_totals result
Clean batch	each delta applied once
Restart mid-batch	reprocessed batch; gate skips applied deltas → no jump
Duplicate delivery	gate rejects → total unchanged
Offset + write	always consistent (same transaction)

Why this works — concept by concept:

• At-least-once + gate = effectively-once — the source stays at-least-once (cheap, lossless); the applied_deltas unique-key gate absorbs duplicates so each delta hits the total once. Exactly-once processing without exactly-once delivery.
• Additive effects need a gate, not just upsert — an idempotent replace (set total = X) is safe under upsert alone, but an additive (total += delta) is not — re-applying adds twice. The dedup gate is mandatory precisely because the effect is additive.
• Offset committed with the write — storing the consumed offset in the same transaction as the effect removes the crash window that caused the original jump. Progress and effect are one atomic unit.
• Checkpoint in the sink DB — putting offsets in Postgres (not Kafka auto-commit) makes "resume position" and "applied effects" share one commit, which is the only way to make them consistent without distributed transactions.
• Cost — one gate row per event (TTL-bounded) plus one upsert; O(1) per event, O(batch) per transaction. Compared to the double-counting bug it replaces, the overhead is a single indexed insert. This is the textbook effectively-once consumer: at-least-once delivery, idempotent processing, checkpoint atomic with output.

Streaming
Topic — streaming
Streaming problems on exactly-once and checkpoints

Practice →

Data Processing Topic — data-processing Data processing problems on stateful streaming

Practice →

---

5. Designing idempotent tasks (Airflow / Spark)

Deterministic task boundaries plus overwrite-by-partition make orchestrated tasks safe to retry, replay, and backfill by construction

The mental model in one line: an orchestrated task is idempotent when it is parameterised by a logical interval (the execution/data interval), reads a source window fixed by that interval rather than the wall clock, and writes by overwriting or upserting the corresponding partition — so that the task's output is a pure function of its interval, and running it once, retrying it, replaying it, or backfilling it all converge to the identical partition contents. In Airflow and Spark this is the difference between a DAG you can safely re-run and one that quietly doubles numbers.

Deterministic boundaries — the first rule.

• Use the data interval, not now(). Airflow supplies data_interval_start / data_interval_end (and the legacy ds) for each run. Filter the source with these, never with now() or CURRENT_DATE, so a run for 2026-09-05 always reads the same window regardless of when it actually executes.
• Why it matters. A task that reads WHERE created_at >= now() - interval '1 day' reads a different window every execution; a retry an hour later reads a shifted window and double-counts the overlap. Backfill is impossible because "yesterday" means something different each time.

Idempotent writes in orchestration.

• Overwrite the interval's partition. Delete-insert (section 3) keyed on data_interval_start, or upsert by natural key. The partition for the interval is fully replaced, so a re-run of the interval yields the same partition.
• Never append. An appending task under an orchestrator with retries is the double-count bug from section 1. If the sink is append-only (e.g. an event log), move dedup to the reader.

Airflow-specific levers.

• retries + retry_delay. Configure retries knowing each attempt is idempotent — otherwise retries are the cause of duplicates, not the cure.
• depends_on_past / wait_for_downstream. Use when an interval's correctness depends on the prior interval's completion (e.g. cumulative state); otherwise leave off so backfills can parallelise.
• catchup and backfill. With deterministic intervals and overwrite writes, catchup=True (or a manual backfill) safely fills history — each interval is an independent idempotent unit.
• Deferred/atomic publish. Write to a staging location, then atomically swap/rename into the final partition, so a partially-written partition is never visible to readers.

Spark-specific levers.

• insertInto with spark.sql.sources.partitionOverwriteMode=dynamic (or replaceWhere on Delta) overwrites only the partitions present in the output, giving delete-insert semantics without a manual delete.
• Deterministic output paths. Write to .../date=<data_interval_start>/; a re-run overwrites the same path. Avoid embedding now() or a random run id in the path.
• _SUCCESS markers + atomic rename. Spark writes to a temp dir and renames on success; treat the _SUCCESS marker as the commit signal so downstream ignores partial output.
• Idempotent foreachBatch. As in section 4, MERGE or overwrite by key so a replayed batch is a no-op.

Common interview probes on idempotent orchestration.

• "How do you make an Airflow task safe to retry?" — deterministic interval boundary + overwrite/upsert the interval's partition.
• "Why not now() in the query?" — non-deterministic window; retries/backfills read different data.
• "How does Spark overwrite only one partition?" — dynamic partition overwrite / replaceWhere, deterministic path.
• "How do you make backfill safe?" — each interval is an independent idempotent unit; overwrite its partition.

Worked example — an Airflow task that is safe to retry and backfill

Detailed explanation. The canonical idempotent Airflow task takes the run's data interval, filters the source to that window, and delete-inserts the corresponding partition. Because everything is keyed on the interval, retries, manual clears, and backfills all produce the same output. Walk through the task.

• Boundary. data_interval_start supplied by Airflow.
• Write. Delete-insert the partition for that interval, in one transaction.
• Retry/backfill. Both re-run the identical delete-insert → identical partition.

Question. Write an Airflow task that idempotently builds one day of daily_revenue, safe to retry and backfill.

Input.

Parameter	Value
Boundary	data_interval_start (e.g. 2026-09-05)
Target	daily_revenue partitioned by revenue_date
Write	delete-insert in one transaction
retries	3

Code.

from airflow.decorators import dag, task
from datetime import datetime
import psycopg2

@dag(schedule="@daily", start_date=datetime(2026, 1, 1),
     catchup=True, default_args={"retries": 3})
def daily_revenue_dag():

    @task
    def build_daily_revenue(data_interval_start=None):
        ds = data_interval_start.date().isoformat()      # logical interval, not now()
        conn = psycopg2.connect("host=warehouse dbname=analytics")
        with conn:                                        # atomic delete + insert
            with conn.cursor() as cur:
                cur.execute(
                    "DELETE FROM daily_revenue WHERE revenue_date = %s", (ds,))
                cur.execute("""
                    INSERT INTO daily_revenue (revenue_date, customer_id, revenue_cents)
                    SELECT %s, customer_id, SUM(total_cents)
                    FROM   fact_orders
                    WHERE  order_ts >= %s::date
                      AND  order_ts <  %s::date + INTERVAL '1 day'
                    GROUP  BY customer_id
                """, (ds, ds, ds))

    build_daily_revenue()

daily_revenue_dag()

Step-by-step explanation.

1. data_interval_start is injected by Airflow for each run; deriving ds from it (not now()) fixes the source window so every execution of the 2026-09-05 run reads the same [2026-09-05, 2026-09-06).
2. The delete-insert is wrapped in with conn: so the two statements commit atomically — a retry that dies mid-task rolls back to the previous good partition.
3. DELETE ... WHERE revenue_date = %s clears the day's partition; on the first run it is a no-op, on a retry it removes the prior attempt's rows. This is what makes re-runs converge.
4. retries=3 is now safe: each attempt performs the identical delete-insert, so at most one attempt's output survives and it equals a clean run. Without idempotency, retries=3 would be three chances to duplicate.
5. catchup=True backfills every missing day since start_date; each day is an independent idempotent unit, so the backfill is just N safe delete-inserts. A manual "clear and re-run" of any day is equally safe.

Output.

Trigger	daily_revenue for 2026-09-05
Scheduled run	correct aggregate
Retry after failure	same aggregate
Manual clear + re-run	same aggregate
Backfill	same aggregate

Rule of thumb. Parameterise every Airflow task on its data interval and make it overwrite (delete-insert or upsert) the interval's partition. Once that holds, turn on retries and catchup freely — they can only converge to the correct result, never duplicate it.

Worked example — idempotent Spark partition overwrite

Detailed explanation. In Spark, the idempotent equivalent of delete-insert is dynamic partition overwrite: write a DataFrame partitioned by the interval, and Spark replaces only the partitions present in the output, leaving other partitions untouched. Combined with a deterministic output path, a re-run overwrites exactly the interval it computed. Walk through the write.

• Mode. spark.sql.sources.partitionOverwriteMode = dynamic.
• Write. insertInto (or .mode("overwrite") with partitionBy) so only the computed partitions are replaced.
• Path. Derived from the data interval; no now() or random id.

Question. Write a Spark job that idempotently overwrites one day's partition of a partitioned table.

Input.

Parameter	Value
Interval	2026-09-05 (passed in)
Table	sales partitioned by sale_date
Mode	dynamic partition overwrite

Code.

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
         .getOrCreate())

ds = "2026-09-05"                              # logical interval, injected by the orchestrator

daily = (spark.read.table("staging_orders")
         .where((F.col("order_ts") >= F.lit(ds).cast("date")) &
                (F.col("order_ts") <  F.date_add(F.lit(ds).cast("date"), 1)))
         .withColumn("sale_date", F.lit(ds).cast("date"))
         .groupBy("sale_date", "customer_id")
         .agg(F.sum("amount_cents").alias("amount_cents")))

# Dynamic overwrite: replaces ONLY the sale_date=2026-09-05 partition, idempotently
(daily.write
      .mode("overwrite")
      .partitionBy("sale_date")
      .format("parquet")
      .insertInto("sales"))

Step-by-step explanation.

1. partitionOverwriteMode = dynamic tells Spark to overwrite only the partitions that appear in the output DataFrame, not the entire table. Without it, overwrite would wipe all partitions — a classic data-loss footgun.
2. The source is filtered to the interval [ds, ds+1) using ds, not now(), so the computed rows are a deterministic function of the day.
3. withColumn("sale_date", lit(ds)) stamps the partition column so every output row belongs to the 2026-09-05 partition.
4. write.mode("overwrite").partitionBy("sale_date").insertInto("sales") replaces exactly the sale_date=2026-09-05 partition with the freshly computed rows. Other days are untouched.
5. A re-run (retry or backfill of that day) recomputes the same rows and overwrites the same partition — idempotent. Because the partition is fully replaced, rows that vanished from the source for that day are also dropped, exactly like the SQL delete-insert.

Output.

Run	sale_date=2026-09-05 partition	Other partitions
First	written	untouched
Retry / backfill	overwritten (same contents)	untouched
Source row removed	absent after overwrite	untouched

Rule of thumb. For Spark writes to partitioned tables, always set partitionOverwriteMode=dynamic, stamp the partition column from the logical interval, and use overwrite. That gives delete-insert semantics per partition with no manual delete and no risk of wiping neighbouring partitions.

Worked example — a poison-pill retry with a dedup gate

Detailed explanation. Not all tasks map cleanly to a partition — some perform per-record side effects (calling an external API, sending a notification) where a retry must not repeat the effect for records already handled. Here the idempotency mechanism is a per-record dedup gate that persists progress within the task, so a retry resumes rather than restarts. Walk through a partially-completed task that retries.

• The task. For each record in a batch, call an external API (a non-transactional side effect).
• The gate. Record each handled record's key in a dedup table before the effect is externally visible; skip handled keys on retry.
• The retry. Resumes from where it failed; already-handled records are skipped.

Question. Make a per-record side-effect task safe to retry so that a failure partway through does not repeat effects for already-processed records.

Input.

Record key	First attempt	After crash	Retry
r1	handled	—	skipped
r2	handled	—	skipped
r3	crash before handling	—	handled

Code.

import psycopg2
from psycopg2 import errors as pg_errors

def process_batch(conn, records, call_external_api):
    for rec in records:
        key = record_key(rec)
        try:
            with conn:                              # per-record transaction
                with conn.cursor() as cur:
                    # 1. Claim the record (gate). Fails if already handled.
                    cur.execute(
                        "INSERT INTO handled_records (rec_key) VALUES (%s)", (key,))
                    # 2. Idempotent external call — pass the SAME key downstream
                    call_external_api(rec, idempotency_key=key)
        except pg_errors.UniqueViolation:
            conn.rollback()
            continue                                # already handled on a prior attempt

Step-by-step explanation.

1. Each record is processed in its own transaction, so progress is durable per record — a crash loses at most the in-flight record, not the whole batch's progress.
2. Step 1 claims the record by inserting its key into handled_records. On the first attempt this succeeds; on a retry for an already-handled record it raises UniqueViolation, which we catch and skip.
3. Step 2 makes the external call, passing idempotency_key=key. A well-behaved API honours this header so that even if step 1 committed but the process died before step 2 fully registered, retrying the call with the same key does not double-apply on the API side.
4. On retry, r1 and r2 are already in handled_records → UniqueViolation → skipped. r3 (which crashed before its claim committed) is unhandled → processed now. The batch completes with each record handled exactly once.
5. The combination — local gate for "did I start this?" plus a downstream idempotency key for "did the API apply this?" — makes a non-transactional side effect effectively-once across retries. Neither layer alone is sufficient for an external effect.

Output.

Record	Effects across attempt + retry
r1	1 (skipped on retry)
r2	1 (skipped on retry)
r3	1 (handled on retry)

Rule of thumb. For per-record external side effects, gate each record with a durable "claim" and propagate an idempotency key to the downstream so it can dedupe too. A retry then resumes rather than repeats, and the external effect fires exactly once even though the task ran twice.

Data engineering interview question on idempotent orchestration

A senior interviewer might ask: "Design a daily aggregation DAG in Airflow that reads an at-least-once staging table and writes a partitioned warehouse fact. It must be safe to retry a failed day, safe to backfill a range of past days in parallel, and it must correctly reflect rows deleted upstream. Walk me through the task boundary, the write strategy, the retry/backfill configuration, and how you'd prove no duplicates after an arbitrary sequence of runs."

Solution Using interval-keyed tasks + transactional delete-insert + a duplicate-key assertion

from airflow.decorators import dag, task
from datetime import datetime
import psycopg2

@dag(schedule="@daily", start_date=datetime(2026, 1, 1),
     catchup=True, max_active_runs=8,          # backfill days in parallel...
     default_args={"retries": 3})
def daily_fact_dag():

    @task
    def load_partition(data_interval_start=None):
        ds = data_interval_start.date().isoformat()
        conn = psycopg2.connect("host=warehouse dbname=analytics")
        with conn:                              # atomic: delete + deduped insert
            with conn.cursor() as cur:
                cur.execute("DELETE FROM fact_orders WHERE order_date = %s", (ds,))
                cur.execute("""
                    INSERT INTO fact_orders (order_date, order_id, customer_id, amount_cents)
                    SELECT %s, order_id, customer_id, amount_cents
                    FROM (
                        SELECT order_id, customer_id, amount_cents, order_ts,
                               ROW_NUMBER() OVER (PARTITION BY order_id
                                                  ORDER BY order_ts DESC) AS rn
                        FROM   staging_orders
                        WHERE  order_ts >= %s::date
                          AND  order_ts <  %s::date + INTERVAL '1 day'
                    ) d
                    WHERE rn = 1
                """, (ds, ds, ds))

    @task
    def assert_no_duplicates(data_interval_start=None):
        ds = data_interval_start.date().isoformat()
        conn = psycopg2.connect("host=warehouse dbname=analytics")
        with conn.cursor() as cur:
            cur.execute("""
                SELECT COUNT(*) FROM (
                    SELECT order_id FROM fact_orders WHERE order_date = %s
                    GROUP BY order_id HAVING COUNT(*) > 1
                ) dups
            """, (ds,))
            dup_keys = cur.fetchone()[0]
        if dup_keys:
            raise ValueError(f"{dup_keys} duplicate order_id(s) in partition {ds}")

    load_partition() >> assert_no_duplicates()

daily_fact_dag()

Step-by-step trace.

Layer	Component	Purpose
Boundary	data_interval_start → ds	deterministic window; not now()
Source dedup	ROW_NUMBER() = 1	collapse at-least-once staging duplicates
Write	DELETE + INSERT in one txn	idempotent partition replace; handles deletes
Parallelism	max_active_runs=8	backfill many days concurrently; each independent
Retry	retries=3	safe — each attempt is the same delete-insert
Assertion	assert_no_duplicates downstream	fails the run if any duplicate key survives

After deployment, retrying a failed day re-runs the delete-insert → identical partition; backfilling a range runs up to 8 days in parallel, each an independent idempotent unit; upstream-deleted rows vanish because the partition is fully replaced; and the assert_no_duplicates task turns idempotency from a claim into a checked invariant that fails loudly if ever violated. An arbitrary sequence of retries, replays, and backfills over any set of days converges to one partition per day with one row per order_id.

Output:

Metric	Value
Duplicate keys after any run sequence	0 (asserted)
Safe parallel backfill	yes (max_active_runs=8)
Upstream deletions reflected	yes (full partition replace)
Retry double-count	impossible (delete-insert)
Idempotency verification	automated per run

Why this works — concept by concept:

• Interval-keyed boundary — deriving the window from data_interval_start makes each day's output a pure function of the day, so retries, replays, and parallel backfills all target the same deterministic window rather than a moving now()-based one.
• Source dedup — ROW_NUMBER() = 1 collapses the at-least-once staging duplicates to one row per key before the insert, so the partition holds exactly one row per order_id regardless of upstream repeats.
• Transactional delete-insert — replacing the partition atomically handles new, changed, and deleted rows in one move, and a crash rolls back to the prior good partition; there is no half-written state and no append to double-count.
• Independent partitions enable parallelism — because each day's task only touches its own partition, max_active_runs=8 backfills days concurrently with no cross-day interference, and depends_on_past is unnecessary for a non-cumulative fact.
• Cost — O(rows per day) for the delete + deduped insert, plus an O(rows per day) assertion query, per run. Compared to a naive append it adds one cheap per-partition delete and one check; in return it makes retry, replay, and backfill provably duplicate-free. The assertion converts "we believe it's idempotent" into a DAG that proves it on every execution.

ETL
Topic — etl
ETL problems on retry-safe orchestration

Practice →

Data Processing
Topic — data-processing
Data processing problems on Spark partition writes

Practice →

---

Cheat sheet — idempotent pipeline recipes

• The invariant. A pipeline operation is idempotent when applying it once and N times leave the same final state. Because retries are inevitable (OOM kills, lost acks, rebalances, scheduler retries), idempotency is the contract that makes retries, replays, and backfills safe. Name the unit of work first; idempotency is always relative to a unit.
• Idempotency key derivation. Prefer a natural/business key (order_id, payment_id, (user_id, event_ts, event_type)). No natural key → content hash: sha256(json.dumps(business_fields, sort_keys=True, separators=(",",":"))), excluding volatile transport metadata (_recv_ts, _attempt, offset). Request/response → a producer-supplied Idempotency-Key generated once per request. Never uuid4() per attempt — the retry gets a new key and dedup never fires.
• Postgres unique-index dedup gate. CREATE TABLE processed_events(event_key TEXT PRIMARY KEY, processed_at TIMESTAMPTZ); inside one transaction INSERT ... event_key then do the side effect; catch UniqueViolation → skip. Gate + effect commit atomically, so "seen" ⇔ "done" with no crash window. TTL the gate to the redelivery window.
• Redis SETNX dedup gate. r.set(f"dedup:{key}", 1, nx=True, ex=TTL) returns truthy only for the first setter. Fast, bounded by TTL. The TTL is a correctness parameter — size it ≥ the max redelivery/replay window (Kafka retention + retry horizon). Always back it with an idempotent sink for the non-transactional edge cases.
• Postgres upsert. INSERT ... VALUES ... ON CONFLICT (key) DO UPDATE SET col = EXCLUDED.col. Requires a real unique key = the idempotency contract. Idempotent for replace effects; not idempotent for additive effects (total += x) — those need a dedup gate too.
• Partition overwrite (delete-insert). BEGIN; DELETE FROM fact WHERE part = :ds; INSERT INTO fact SELECT ... WHERE <window(:ds)>; COMMIT;. Simpler than MERGE, handles disappeared rows for free, partition is the idempotency unit. Dedupe the source (ROW_NUMBER() = 1 latest) if it is at-least-once.
• Warehouse MERGE by key. Dedupe the source to one row per key (QUALIFY ROW_NUMBER() OVER (PARTITION BY k ORDER BY ts DESC) = 1), then MERGE ... ON t.k = s.k WHEN MATCHED UPDATE ... WHEN NOT MATCHED INSERT .... Add WHEN NOT MATCHED BY SOURCE THEN DELETE only for full snapshots, not incremental deltas.
• Exactly-once decomposition. exactly-once = at-least-once delivery + idempotent processing (+ checkpoint atomic with output). Exactly-once delivery is impossible (two-generals); exactly-once effect is achievable. The three things that must be atomic: output write, state checkpoint, source-offset advance.
• Spark Structured Streaming. Stable checkpointLocation (offsets + state) + idempotent foreachBatch (MERGE by key or dynamic overwrite). A failed batch replays with identical offsets → the idempotent sink makes the replay a no-op.
• Kafka effectively-once. Transactional producer (enable.idempotence=true, stable transactional.id), send_offsets_to_transaction, commit_transaction — output + offsets atomic; downstream isolation.level=read_committed so aborted batches are invisible.
• Airflow retry-safe task. Key on data_interval_start (never now()); overwrite/upsert the interval's partition in one transaction; then retries and catchup/backfill are safe by construction. Independent partitions → max_active_runs parallel backfill. Add a HAVING COUNT(*) > 1 assertion task to prove zero duplicate keys.
• Non-deterministic pitfalls. now()/CURRENT_DATE in the source filter (moving window), uuid4() per attempt (breaks dedup), bare INSERT into a keyed table (appends duplicates), committing the offset after the write (crash window). Each reintroduces duplicates under failure.
• Migration cost. Append → upsert: add a unique key + rewrite the write (~hours). Append → delete-insert: add partition column + wrap in a transaction (~hours). Add a dedup gate: one table + a try/except (~hours). Retrofitting idempotency is cheap; a double-counted-revenue incident is not.

Frequently asked questions

What does idempotent mean for a data pipeline?

An operation is idempotent when applying it once and applying it any number of times leave the target system in the same final state. For an idempotent data pipeline this means a task can be retried, replayed, or backfilled without creating duplicate rows, double-counting metrics, or repeating side effects. The property is always defined relative to a unit of work — one event, one batch, one partition, one task instance for a logical interval — so "is this pipeline idempotent?" only becomes answerable once you name the unit. Idempotency matters because failures in distributed systems are inevitable (OOM kills, lost acknowledgements, consumer rebalances, scheduler retries), so the practical question is never "how do I avoid re-running?" but "how do I make re-running safe?"

What is an idempotency key and how do I choose one?

An idempotency key is a deterministic identifier for a unit of work: the same logical input must always produce the same key, on the first attempt and every retry. Choose it in priority order — first a natural/business key that already uniquely identifies the record (order_id, payment_id, or a composite like (user_id, event_ts, event_type)); if none exists, a content hash (sha256 over the canonicalised business fields, with volatile transport metadata excluded); and for request/response systems, a client-generated Idempotency-Key created once per request. The cardinal mistake is generating a fresh uuid4() inside the retry path — every attempt then gets a different key, so the deduplication store never recognises the retry and admits every copy.

Is exactly-once delivery real?

No — exactly-once delivery is impossible across an unreliable network, because a sender can never be certain its message arrived (the acknowledgement itself can be lost), so it must choose between at-most-once (risk loss) and at-least-once (risk duplicates). What streaming engines market as "exactly-once" is really effectively-once processing: at-least-once delivery combined with idempotent processing (dedup or upsert) and checkpointed progress that commits atomically with the output. The duplicates that at-least-once delivery produces are absorbed by the idempotent processing layer, so the observable effect is one-per-event even though the wire saw duplicates. The equation to remember: exactly-once = at-least-once + idempotent processing + checkpointing.

Upsert vs delete-insert — which for idempotency?

Both make a sink idempotent; pick by the unit of work. Use upsert (INSERT ... ON CONFLICT DO UPDATE or MERGE) when the unit is a set of rows identified by natural keys and you want in-place updates — dimensions, late-arriving corrections, keyed facts. Use delete-insert (delete the partition, then insert the recomputation, in one transaction) when the unit is a whole partition such as a day or an hour: it is simpler than MERGE, and it removes rows that disappeared from the source for free, because the entire partition is replaced. A naive MERGE leaves stale target rows behind unless you add WHEN NOT MATCHED BY SOURCE THEN DELETE, so for full-partition recomputes delete-insert is usually the cleaner mental model. One caveat: upsert alone is idempotent for replace effects but not for additive ones (total += delta), which also need a dedup gate.

How do I make an Airflow task safe to retry?

Two rules. First, parameterise the task on its logical interval — use data_interval_start / data_interval_end (or ds), never now() or CURRENT_DATE, so a run for a given day always reads the same source window regardless of when it actually executes. Second, make the write overwrite the interval's partition rather than append: delete-insert the partition in one transaction, or upsert by natural key. Once both hold, the task's output is a pure function of its interval, so retries can only converge to the correct result, and catchup/backfill simply runs each interval as an independent idempotent unit. Add a downstream assertion task (GROUP BY key HAVING COUNT(*) > 1) to fail loudly if a duplicate key ever slips through, turning idempotency from a claim into a checked invariant.

How does checkpointing prevent duplicates?

checkpointing records processing progress — source offsets and any accumulated state — so a job can resume after a crash without skipping input or reprocessing from scratch. It prevents duplicates only when the checkpoint commits atomically with the output write. The classic bug is writing output rows and then committing the offset separately: a crash in that window replays the batch and re-applies it. The fix is to make the offset advance and the effect share one commit — store the offset in the same transactional sink as the write (Postgres), or use the engine's transaction (Kafka's send_offsets_to_transaction, Spark's checkpoint tied to batch completion, Flink's two-phase commit). Checkpointing plus an idempotent sink is what turns at-least-once delivery into effectively-once processing across restarts.

Practice on PipeCode

• Drill the ETL practice library → for the retry-safe load, delete-insert, upsert, and backfill problems senior interviewers love.
• Rehearse on the data processing practice library → for deduplication, idempotency-key derivation, and partition-overwrite patterns.
• Sharpen the streaming axis with the streaming practice library → for at-least-once dedup, exactly-once, and checkpointing scenarios.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the idempotency checklist against real graded inputs.

Lock in idempotency muscle memory

Docs explain the concept. PipeCode drills explain the decision — when a bare INSERT double-counts, when a content hash beats a natural key, when upsert is not enough for an additive effect, when the offset must commit with the write. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs data engineers actually face.

Practice ETL problems →
Practice streaming problems →