Idempotent Writes, and Why Retries Duplicate Data

A load step fails at 01:47. The orchestrator does what it was configured to do: waits thirty seconds and tries again. The retry succeeds, the run goes green, and nobody is woken up. At 09:00 finance asks why yesterday’s revenue is exactly double.

Nothing was broken, and that’s the uncomfortable part. The first attempt did write its rows. What failed was everything after the write: the acknowledgment timed out, the status call returned a 503, the driver was evicted a second after the sink accepted the data. The orchestrator saw a failed step. The table saw a successful one. The retry wrote everything a second time.

This is the most common way a healthy-looking pipeline produces wrong numbers. Fixing it doesn’t need a new tool. It needs the write to be idempotent.

The definition that matters

An operation is idempotent if doing it once and doing it N times leave the system in the same state. For a pipeline, that means running the job twice for the same input produces the same table as running it once.

That property matters because every layer of a data platform will re-execute work at some point:

• Orchestrators retry failed activities. Fabric and ADF pipelines, Airflow, and Dagster all have per-step retry policies.
• Spark re-runs failed tasks, and with speculative execution it launches duplicate copies of slow ones.
• Streaming jobs restart from their last checkpoint and replay any micro-batch that wasn’t recorded as complete.
• Kafka consumers re-read anything after the last committed offset.
• People rerun jobs and backfill date ranges.

None of these systems promises to run your code exactly once. What they offer is at-least-once execution. “Exactly-once” isn’t a feature you can switch on in the transport. It’s something you build at the sink.

Where the second copy comes from

It helps to list the mechanisms that re-execute work, because each one replays a different unit and breaks a different kind of write.

Re-execution source	What gets replayed	Duplicates when the write is…
Orchestrator retry	The whole activity	An append (INSERT INTO, copy activity in append mode)
Spark task retry / speculation	One task’s partition of output	Non-transactional: JDBC, REST calls, files written outside the commit protocol
Streaming restart	The last uncommitted micro-batch	A plain append inside foreachBatch
Manual rerun or backfill	A date range	Anything keyed on “rows I haven’t seen”, or plain appends
Upstream redelivery	Events you already have	Trusting the producer never to resend

The Spark row is the subtle one. Files written through DataFrameWriter into a Delta table go through a commit protocol: a failed task’s output is never committed, so task retries are safe. The JDBC writer is different. It commits one database transaction per partition, so a task that commits and then loses its executor before reporting success gets rescheduled, and its rows go in a second time. Anything you do inside foreachPartition, like calling an API or writing to a queue, has the same exposure.

Four patterns that make a write safe to repeat

1. Replace the unit of work

If the job’s logical unit is one day of data, make the write replace that day instead of adding to it. Rerunning then replaces the previous attempt’s output with an identical copy.

def load_day(spark, logical_date: str):
    day = (
        spark.read.table("bronze.orders")
        .where(f"order_date = '{logical_date}'")
        .transform(clean_orders)          # deterministic: same input -> same rows
    )

    (day.write.format("delta")
        .mode("overwrite")
        .option("replaceWhere", f"order_date = '{logical_date}'")
        .saveAsTable("silver.orders"))

replaceWhere atomically deletes the rows matching the predicate and writes the new ones in a single Delta commit. Readers see either the old day or the new day, never half of each. Since Delta 2.0 the predicate can be on any column, not only partition columns, and by default Delta rejects the write if any incoming row falls outside the predicate.

This pattern requires one thing: the job must recompute the whole unit from its source every time. If it computes “only rows since my last cursor”, replacing the day with that partial set deletes data. That leads to the second pattern.

2. Upsert on a natural key

When the unit of work isn’t a clean slice, as with CDC feeds, API extracts, or event streams with updates, key the write on something that identifies the real-world entity and MERGE on it.

MERGE INTO silver.orders AS t
USING (
  SELECT *
  FROM (
    SELECT *,
           ROW_NUMBER() OVER (
             PARTITION BY order_id
             ORDER BY source_updated_at DESC, source_lsn DESC
           ) AS rn
    FROM bronze.orders_changes
    WHERE batch_date = '2026-10-01'
  )
  WHERE rn = 1
) AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s.source_updated_at > t.source_updated_at THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *

Three details make this idempotent rather than just “an upsert”:

1. The source is deduplicated first. Delta fails a MERGE when more than one source row matches the same target row. Even if it didn’t, the result would depend on which row won. Deduplicate inside the batch with a deterministic ordering.
2. The tiebreak is total. source_updated_at alone can tie. Add a second column that can’t, like a log sequence number or offset, so two runs always choose the same row.
3. The update is guarded. s.source_updated_at > t.source_updated_at means replaying an older batch can’t overwrite newer state. Without the guard, a rerun of last Tuesday quietly rolls today’s rows back to Tuesday’s values.

rn ends up in the table here because of INSERT *. In production, list the columns explicitly. That habit also protects you from schema drift.

3. Let the sink remember what it has already committed

Structured Streaming into a Delta sink is exactly-once on its own: the checkpoint and the Delta log together record which batch IDs have been written. As soon as you use foreachBatch, which you need for MERGE or for writing to several tables, that guarantee goes away. A batch that wrote its data but crashed before the checkpoint recorded it gets replayed after the restart.

Delta gives you a direct fix, idempotent table writes: tag each write with an application ID and a monotonically increasing version, and Delta skips any write whose version it has already seen for that ID.

APP_ID = "bronze_orders_ingest_v1"

def write_batch(batch_df, batch_id: int):
    (batch_df.write.format("delta")
        .mode("append")
        .option("txnAppId", APP_ID)
        .option("txnVersion", batch_id)
        .saveAsTable("bronze.orders"))

(spark.readStream.table("landing.orders_raw")
    .writeStream
    .foreachBatch(write_batch)
    .option("checkpointLocation", "Files/checkpoints/bronze_orders")
    .start())

4. Idempotency keys for side effects

Some writes aren’t to tables: sending an email, posting to a billing API, publishing an event. For these, give each logical operation a deterministic key, derived from the input (for example sha2(concat(order_id, ':', 'invoice'), 256)) and never from uuid() at runtime. Then either:

• send the key to a service that honours it (many payment and messaging APIs accept an Idempotency-Key header), or
• record processed keys in a ledger table and skip any key already present. The ledger insert and the side effect can’t share a transaction across systems, so prefer writing the intent to an outbox table in the same transaction as your data, with a separate sender that is itself keyed.

What quietly breaks idempotence

Most pipelines that duplicate data were meant to be idempotent. They stopped being idempotent through one of these:

• Relative dates in the job body. WHERE order_date = current_date() - 1 processes a different day if the retry happens after midnight. Pass the logical date in from the orchestrator as a parameter and never derive it from the clock inside the job. Event time vs processing time goes deeper on this.
• Generated values used as keys. uuid(), monotonically_increasing_id(), current_timestamp(), and identity columns all produce new values on every run, so a MERGE on them never matches. Ingestion timestamps are fine as attributes. They must never be part of the key.
• Cursors that move before the data lands. If the high-watermark (“last processed updated_at”) is saved before the write commits, a failure between the two skips data. If it’s saved after, in a separate system, a failure between the two duplicates data. Derive the cursor from the target (SELECT max(source_updated_at) FROM silver.orders) or commit it in the same transaction.
• A non-deterministic transform. dropDuplicates() without a defined order, first() over an unordered group, or sampling: the row count stays stable across reruns, but the content doesn’t. That is still a correctness bug.
• Copy activities in append mode. The default behaviour of many low-code copy tools is to append. Write to a staging table that you truncate, then MERGE or replaceWhere from staging.

The test: run it twice

Idempotence is easy to assert and almost nobody does. Run the job, fingerprint the target, run it again with the same parameters, and compare.

def fingerprint(spark, table: str, key: str, measure: str):
    return spark.sql(f"""
        SELECT count(*)              AS row_count,
               count(DISTINCT {key}) AS key_count,
               sum({measure})        AS total
        FROM {table}
    """).first()

def test_orders_load_is_idempotent(spark, seeded_bronze):
    load_day(spark, logical_date="2026-10-01")
    first = fingerprint(spark, "silver.orders", "order_id", "amount")

    load_day(spark, logical_date="2026-10-01")
    second = fingerprint(spark, "silver.orders", "order_id", "amount")

    assert first == second
    assert first.row_count == first.key_count   # one row per order: the grain holds

The last assertion checks something slightly different: that the table has the grain it claims to have. Duplicates from retries and fan-out from bad joins look the same in a dashboard, and the same assertion catches both.

On Monday

• For each job, write down its unit of work and how it’s written: replace, merge, or append. Every append needs a reason.
• Make the logical date a parameter. Remove current_date() from job bodies.
• Make sure every MERGE source is deduplicated with a total ordering and guarded against older data.
• For foreachBatch sinks, add txnAppId and txnVersion, and tie the app ID to the checkpoint.
• Add the run-it-twice test to CI for the tables people make decisions from.

The orchestrator’s retry button is a promise that running the job again is safe. Idempotent writes are what keep that promise.