Your foreachBatch wrote the batch twice — at-least-once is the contract, not a bug

The same batch can arrive twice

foreachBatch promises at-least-once writes — not exactly-once. When a micro-batch fails after its writes commit (or the query restarts mid-batch), the engine re-delivers the same batch_id and your handler runs again from the top. A plain .mode("append") write has no memory of the first delivery, so the retry doubles every row.

The fix is writer-level idempotency: on Delta targets, pass .option("txnAppId", APP_ID).option("txnVersion", batch_id). Delta records the (app, version) pair in the transaction log and ignores a re-delivered batch. For anything outside Delta — an API, a database, or a queue — key your own idempotency on the batch_id: same batch id in, same result out, no matter how many times the engine calls you.

Illustrative simulation

Not a live streaming run: the engine's documented retry is simulated by calling the batch handler twice with the same batch_id — exactly what a re-delivered micro-batch looks like. Row counts below are deterministic from the setup, not measured production figures.

Setup — safe to rerun

spark.sql("""
CREATE OR REPLACE TABLE demo.orders_sink (
  order_id STRING,
  amount INT
) USING DELTA
""")

batch_df = spark.createDataFrame(
    [("ord1", 100), ("ord2", 200), ("ord3", 300)],
    "order_id STRING, amount INT",
)

APP_ID = "orders-streaming-job-v1"

Step 1 — the broken handler

def write_batch_broken(batch_df, batch_id):
    if batch_df.isEmpty():
        return

    (batch_df.write
      .format("delta")
      .mode("append")
      .saveAsTable("demo.orders_sink"))

# Simulate the original delivery and its retry.
write_batch_broken(batch_df, 7)
write_batch_broken(batch_df, 7)

spark.sql("SELECT COUNT(*) AS rows FROM demo.orders_sink").show()

Step 2 — bind the Delta transaction version to the batch id

# Fresh target: this keeps the simulation idempotent and isolates Step 2.
spark.sql("""
CREATE OR REPLACE TABLE demo.orders_sink (
  order_id STRING,
  amount INT
) USING DELTA
""")

APP_ID = "orders-streaming-job-v1"

def write_batch(batch_df, batch_id):
    if batch_df.isEmpty():
        return

    (batch_df.write
      .format("delta")
      .mode("append")
      .option("txnAppId", APP_ID)
      .option("txnVersion", batch_id)
      .saveAsTable("demo.orders_sink"))

write_batch(batch_df, 7)
write_batch(batch_df, 7)

spark.sql("SELECT COUNT(*) AS rows FROM demo.orders_sink").show()

Step 3 — the checkpoint-reset trap

# Fresh target for this isolated reset simulation.
spark.sql("""
CREATE OR REPLACE TABLE demo.orders_sink (
  order_id STRING,
  amount INT
) USING DELTA
""")

APP_ID = "orders-streaming-job-v1"

# Original checkpoint lineage: versions 0, 1, and 2 commit nine rows.
for b in range(3):
    write_batch(batch_df, b)

# A new checkpoint starts batch ids at 0, but APP_ID was mistakenly reused.
new_batch = spark.createDataFrame(
    [("ord9", 900)],
    "order_id STRING, amount INT",
)
write_batch(new_batch, 0)

spark.sql("SELECT COUNT(*) AS rows FROM demo.orders_sink").show()

1. Empty-batch guard first: if batch_df.isEmpty(): return — Delta sources can emit empty micro-batches.
2. Append path: use .option("txnAppId", APP_ID).option("txnVersion", batch_id) — bind the version to the batch id, never a random UUID.
3. Keep APP_ID stable per checkpoint lineage; a new checkpoint gets a new app id.
4. MERGE path: dedupe the source to one row per merge key; the merge must converge to the same state when the same batch replays.
5. External sinks: key idempotency on batch_id — same batch id in, same result out.
6. Errors: fail fast to the orchestrator; catch only transient sink errors (retry or DLQ) or suppress duplicate-key violations on an idempotent sink.
7. One sink per writer for throughput: multiple sinks in one foreachBatch serialize writes.
8. Stay in micro-batch mode: never use foreachBatch with continuous processing mode, and never assume a green micro-batch means the retry is a no-op.

Use this when

1. Delta MERGE from a stream: Databricks documentation requires foreachBatch for streaming merge operations.
2. Sinks without streaming writers: Cassandra, a REST API, custom Delta logic, or a dead-letter-queue split — foreachBatch is the documented bridge, with batch_id-keyed idempotency.
3. Orchestrator-retried batches: Lakeflow Jobs or Airflow can retry while the handler stays simple and fail-fast.
4. You can name the idempotency key: stable APP_ID per checkpoint lineage, with batch_id as the version.

Avoid this when

1. A native streaming sink exists: .toTable() is simpler, parallel, and exactly-once through the checkpoint and Delta transaction log.
2. You need multiple sinks at speed: writes in one foreachBatch serialize; use one streaming writer per sink.
3. You use continuous processing mode: foreachBatch relies on micro-batch execution; use foreach() there.
4. You plan local retry loops with swallowed exceptions: they convert retried batches into silent partial writes.