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

Spark Structured Streaming is exactly-once end to end when the source can be replayed, the checkpoint records what each batch read, and the sink ignores a batch it has already written. Kafka and Delta sources replay by offset, and the built-in Delta sink is idempotent. foreachBatch is only at-least-once: when a job fails after the write but before Spark commits the batch, the restart runs the batch again. Make that harmless by writing to Delta with txnAppId and txnVersion set to the batch id, or with a MERGE on the key.

- Page: https://tecyfy.com/data-engineering/course/streaming/exactly-once
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 9: Streaming, Exactly-once and foreachBatch
- Parts of this module: Kafka into a Delta table (https://tecyfy.com/data-engineering/course/streaming); Event time, windows and watermarks (https://tecyfy.com/data-engineering/course/streaming/event-time); Exactly-once and foreachBatch (https://tecyfy.com/data-engineering/course/streaming/exactly-once); Streaming on Databricks (https://tecyfy.com/data-engineering/course/streaming/on-databricks); Build it: orders from Kafka to an hourly table (https://tecyfy.com/data-engineering/course/streaming/build-it)

At the end of this part you can:

- Explain how offsets, commits and an idempotent sink together give exactly-once, and where a failure leaves a batch to run again
- Show that a plain append in foreachBatch duplicates a batch that runs twice
- Make a foreachBatch write idempotent with txnAppId and txnVersion, or with a MERGE on the key

[Part 1](/data-engineering/course/streaming) read the topic `orders` into `bronze_orders`, and [part 2](/data-engineering/course/streaming/event-time) 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:

```python
"""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:

```bash
python copy_events.py --crash
python copy_events.py
```

```text
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:

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/streaming/exactly-once)

## 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`:

```bash
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
```

```text
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`:

```python
"""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:

```text
+--------+-----------+------+------+-------------------+
|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](/data-engineering/course/getting-data-in/change-data-capture) MERGE, run once per batch; [idempotence](https://tecyfy.com/concepts/idempotent-pipeline) 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](/data-engineering/course/streaming/on-databricks) runs these streams on Databricks, and asks when a stream is worth running at all.

## Check yourself

1. The stream in copy_events.py was failing, so someone deleted checkpoints/events_copy and ran it again with --idempotent and the same txnAppId. events_copy did not get the new events. Why?

   A. Delta was down
   B. A new checkpoint starts again at batch 0, and Delta skips every txnVersion it has already seen for that txnAppId, so the new batches look like old ones
   C. availableNow does not work with a new checkpoint
   D. The topic was empty

2. Which of these writes is safe to run twice for the same batch without changing the result?

   A. batch.write.mode("append").saveAsTable(...)
   B. Inserting each row into Postgres with INSERT
   C. Sending each row to a Kafka topic
   D. A MERGE that inserts new keys and updates a key only when the batch's event is newer

### Answers

1. B. Delta keeps, per txnAppId, the highest txnVersion written, and skips anything not above it. Batch ids restart at 0 with a new checkpoint, so the writes are ignored as duplicates. Databricks' docs say to use a different txnAppId whenever you start a query on a new checkpoint.
2. D. The MERGE leaves a table in the same state however often the same batch is applied, because the second time every key matches and no event is newer. A plain append, a plain INSERT and a Kafka write each add the rows again; Spark's docs list the Kafka sink as at-least-once.

## Questions

### Is Spark Structured Streaming exactly-once?

Yes, end to end, when the source can replay data by offset, such as Kafka or a Delta table, and the sink is idempotent, such as Spark's file sink or Delta's. Spark records each batch's offsets in the checkpoint before running it and commits it after, so a failed batch runs again with the same data. foreachBatch, the Kafka sink and foreach are at-least-once, so their writes must be made idempotent.

### What are txnAppId and txnVersion in Delta Lake?

They are write options that make a Delta write idempotent. txnAppId names the writer and txnVersion is a number that only goes up, such as the batch id in foreachBatch. Delta records the pair with the write and ignores a later write with a txnVersion it has already seen for that txnAppId, so a batch that runs twice is written once.

## Sources

- [Structured Streaming Programming Guide, getting started](https://spark.apache.org/docs/latest/streaming/getting-started.html)
- [Structured Streaming Programming Guide: APIs on DataFrames and Datasets](https://spark.apache.org/docs/latest/streaming/apis-on-dataframes-and-datasets.html)
- [Use foreachBatch to write to arbitrary data sinks](https://docs.databricks.com/aws/en/structured-streaming/foreach)
- [Delta Lake table streaming reads and writes](https://docs.databricks.com/aws/en/structured-streaming/delta-lake)
- [Production considerations for Structured Streaming](https://docs.databricks.com/aws/en/structured-streaming/production)