Exactly-once in Spark Structured Streaming: checkpoints, foreachBatch and idempotent writes

Part 1 read the topic orders into bronze_orders, and part 2 counted it by the hour. To start here, run part 1 with all 4 batches of make_orders.py: start Kafka, create the topic, then send each batch and run python read_orders.py. bronze_orders then has 9 events.

Where exactly-once comes from

A stream fails sometimes: the machine is replaced, a write times out, someone stops the job. What matters is what the restart does. Spark's answer has 3 parts:

1. A source that can replay. Kafka keeps its events, and a Delta table keeps its versions, so the same batch can be read again by its offsets.
2. A checkpoint with 2 logs. Before a batch runs, Spark writes the offsets it will read to offsets/. After the batch is written, it writes commits/. On restart, a batch with offsets but no commit runs again, with exactly the same offsets.
3. A sink that ignores a batch it has seen. Spark's Delta and file sinks record which batch they wrote, so the batch that runs again writes nothing new.

All 3 together are what Spark calls end-to-end exactly-once: each event affects the output once, however often a batch runs. The streams in parts 1 and 2 had all 3. The third is the one you lose when you write the batch yourself.

foreachBatch writes at least once

foreachBatch hands each batch to your function as an ordinary DataFrame, so you can write it anywhere, with any code: a MERGE, 2 tables, a database. Spark cannot know what that code did, so Spark's docs say it is only at-least-once. Save this as copy_events.py, which copies bronze_orders to events_copy, and with --crash stops each batch after its write, as a job that dies at the wrong moment would:

"""Copy bronze_orders into events_copy with foreachBatch. --crash fails each batch after its write;
--idempotent tags each write with the batch's id, so Delta skips a batch it has already written."""
import re
import sys
from stream_session import stream_spark

CRASH = "--crash" in sys.argv
IDEMPOTENT = "--idempotent" in sys.argv
spark = stream_spark("copy_events")


def write_batch(batch, batch_id):
    writer = batch.write.format("delta").mode("append")
    if IDEMPOTENT:
        writer = writer.option("txnAppId", "copy_events").option("txnVersion", batch_id)
    writer.saveAsTable("events_copy")
    if CRASH:
        raise RuntimeError(f"stopped after writing batch {batch_id}, before it was committed")


query = (
    spark.readStream.table("bronze_orders")
    .writeStream.foreachBatch(write_batch)
    .option("checkpointLocation", "checkpoints/events_copy")
    .trigger(availableNow=True)
    .start()
)
try:
    query.awaitTermination()
except Exception as error:
    reason = re.search(r"stopped after writing batch \d+, before it was committed", str(error))
    print("the stream failed:", reason.group(0) if reason else type(error).__name__)

copy = spark.table("events_copy")
print(f"events_copy has {copy.count()} rows, {copy.distinct().count()} different")

Run it with --crash, then without:

python copy_events.py --crash
python copy_events.py

the stream failed: stopped after writing batch 0, before it was committed
events_copy has 9 rows, 9 different
events_copy has 18 rows, 9 different

The first run wrote the 9 events and stopped before the commit; checkpoints/events_copy has offsets/0 but no commits/0. So the restart ran batch 0 again, as it should, and the plain append wrote its 9 rows a second time. Nothing failed and nothing warned: the table just counts every order twice. Try the other moments a job can stop at, and the 2 kinds of write:

batch.write.format("delta").mode("append").saveAsTable("events_copy")

Tag the write with the batch

Delta has 2 write options for this. txnAppId names the writer and txnVersion is a number that only grows; Delta records the pair with each write and skips a write whose txnVersion it has already seen for that txnAppId. The batch id is that number. Clear the copy and try again with --idempotent:

rm -r checkpoints/events_copy
python -c "from stream_session import stream_spark; stream_spark().sql('DROP TABLE events_copy')"
python copy_events.py --crash --idempotent
python copy_events.py --idempotent

the stream failed: stopped after writing batch 0, before it was committed
events_copy has 9 rows, 9 different
events_copy has 9 rows, 9 different

Batch 0 ran twice, and was written once. Databricks' docs say to use these options for every Delta write in foreachBatch, and to let an error stop the batch rather than catching it and retrying inside the function, so Spark retries the whole batch. They also warn about the one trap: a new checkpoint starts again at batch 0, so if you ever delete one, give the stream a new txnAppId, or Delta will skip its batches as old ones.

Or write in a way that can run twice

The other fix is a write whose result does not change when it is repeated. A MERGE on the key is one. Silver should hold each order's newest state, not every event, which is a MERGE anyway. Save this as latest_orders.py:

"""Keep the newest state of each order in the Delta table orders, by MERGE in foreachBatch.
--crash fails each batch after its MERGE, before Spark commits it."""
import sys
from pyspark.sql import functions as F
from stream_session import stream_spark

CRASH = "--crash" in sys.argv
spark = stream_spark("latest_orders")
spark.sql("""
CREATE TABLE IF NOT EXISTS orders (order_id INT, customer_id STRING, status STRING,
                                   amount DECIMAL(10,2), event_time TIMESTAMP) USING delta
""")


def merge_batch(batch, batch_id):
    # A batch can hold 2 events for 1 order: keep its newest.
    newest = (
        batch.groupBy("order_id")
        .agg(F.max_by(F.struct("customer_id", "status", "amount", "event_time"), "event_time").alias("event"))
        .select("order_id", "event.*")
    )
    newest.createOrReplaceTempView("newest_events")
    batch.sparkSession.sql("""
        MERGE INTO orders t
        USING newest_events s
        ON t.order_id = s.order_id
        WHEN MATCHED AND s.event_time > t.event_time THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)
    if CRASH:
        raise RuntimeError(f"stopped after merging batch {batch_id}, before it was committed")


query = (
    spark.readStream.table("bronze_orders")
    .writeStream.foreachBatch(merge_batch)
    .option("checkpointLocation", "checkpoints/orders")
    .trigger(availableNow=True)
    .start()
)
try:
    query.awaitTermination()
except Exception as error:
    print("the stream failed:", "stopped after merging" if "stopped after merging" in str(error) else type(error).__name__)

spark.table("orders").orderBy("order_id").show()

python latest_orders.py --crash prints the stream failed: stopped after merging, then the table; python latest_orders.py prints the same table again:

+--------+-----------+------+------+-------------------+
|order_id|customer_id|status|amount|         event_time|
+--------+-----------+------+------+-------------------+
|    1001|        C01|  paid| 25.50|2026-10-01 10:05:00|
|    1002|        C02|placed| 12.00|2026-10-01 09:40:00|
|    1003|        C03|placed| 31.00|2026-10-01 09:55:00|
|    1004|        C01|placed| 40.00|2026-10-01 10:20:00|
|    1005|        C02|placed| 18.25|2026-10-01 11:30:00|
|    1006|        C03|placed| 22.00|2026-10-01 09:50:00|
|    1007|        C02|placed| 15.00|2026-10-01 10:45:00|
|    1008|        C01|placed| 12.00|2026-10-01 12:40:00|
+--------+-----------+------+------+-------------------+

The second run merged batch 0 again: every order matched, none had a newer event, and nothing changed. The condition s.event_time > t.event_time does 2 jobs: a replay changes nothing, and an old event arriving late cannot overwrite a newer state. Order 1001 shows paid, its newest event. This is module 6's MERGE, run once per batch; idempotenceIdempotent Pipelines and Exactly-OnceA pipeline step is idempotent when running it twice gives the same result as running it once. That is what makes retries safe, and it is how systems that deliver a message more than once still count it exactly once. is the property that makes it safe.

Outside Delta

The same reasoning holds for every sink. Spark's docs list Kafka and foreach as at-least-once, so a stream that writes to a topic or calls an API can deliver a record twice, and whoever reads it should drop repeats by a key. For a database, write with an upsert on the key, such as Postgres's INSERT ... ON CONFLICT. A plain insert or a plain append is the one write that a replay always breaks.

Part 4 runs these streams on Databricks, and asks when a stream is worth running at all.