Exactly-once is won in the sink.

01 · THE FAILURE

The pipeline was green while rows disappeared.

My first sink partitioned output by event date and overwrote the partition touched by each micro-batch. A one-batch unit test passed. A running stream exposed the mistake: every batch for today wrote to the same date partition, so each new write removed the earlier one.

The count made the problem undeniable: 3,091 records entered the pipeline and only 926 remained on disk. This was data loss caused by a write strategy that looked reasonable in isolation.

02 · THE DECISION

Give every micro-batch an idempotent destination.

A checkpoint records which source offsets Spark has processed, but a failure can still cause the last batch to run again. The sink therefore has to make a repeated write harmless.

I changed the path so each batch writes beneath its own batch=<id> prefix. Replaying batch 7 overwrites only batch 7. A later batch gets a different prefix, so it cannot erase the earlier output. Event date remains a nested partition for pruning.

target = Path(path) / f"batch={batch_id}"

(
    frame.write.mode("overwrite")
    .partitionBy("event_date")
    .parquet(str(target))
)

03 · THE PROOF

Test the recovery behaviour, not the setting.

The important assertion is not that a configuration flag exists. It is that the stored data stays correct through the failures the system is designed to survive.

• Write the same batch twice and confirm the row count does not increase.
• Write later batches on the same event date and confirm earlier rows remain.
• Send valid and malformed records through the fan-out writer, replay the batch, and confirm both destinations remain idempotent.
• Run several successive batches and verify that all of them survive.