How does a Spark Structured Streaming query achieve end-to-end exactly-once output?

## What "exactly-once" means here The phrase is routinely misheard. Spark does **not** promise that a record is read once or that a task runs once — tasks are retried, whole micro-batches are re-executed after a driver failure, and a record can be processed several times. What Structured Streaming offers is *exactly-once semantics with respect to the sink*: the observable output is as if every input record contributed exactly once. Getting there is a joint effort between the engine, the source and the sink, and the engine only owns one third of it. ## The checkpoint directory `option("checkpointLocation", path)` is mandatory for any production query, and the directory it names is the query's durable identity. Its contents: - **`offsets/`** — the write-ahead log. Before micro-batch *N* processes anything, Spark writes `offsets/N` describing the exact input range for that batch (for Kafka, a per-partition start/end offset map). This is written *ahead* of processing, which is the whole point. - **`commits/`** — one file per batch, written only after the batch's output has been successfully committed to the sink. - **`metadata`** — the query's persistent ID. - **`state/`** — the state store data for stateful operators, partitioned and versioned per batch. - **`sources/`** — source-specific bookkeeping such as the file source's seen-files log. On restart, Spark reads the last `offsets` entry. If there is no matching `commits` entry, that batch is re-planned over the *identical* offset range and re-executed. Determinism of the range is what makes the replay safe: batch 412 always covers the same records, so a sink that can recognise or overwrite batch 412's output ends up in the same state whether the batch ran once or three times. Because the checkpoint is the query's identity, two queries must never share a directory, and deleting it restarts the query from `startingOffsets` with empty state — which for a stateful pipeline means recomputing or losing history. ## What the source must provide Replay only works if the source can be asked again for a byte-identical range. Kafka can (offsets are stable per partition). The file source can (its log tracks which files were seen). Delta and other versioned table sources can. A socket source or a rate source cannot, and Spark documents them as test-only for exactly this reason. One consequence that surprises people: for the Kafka source, offsets live in **Spark's checkpoint**, not in Kafka's `__consumer_offsets`. Spark assigns partitions directly rather than joining a consumer group, so Kafka-side consumer-group lag tooling shows nothing for a Structured Streaming query. `startingOffsets` only applies when the checkpoint is empty; on any subsequent start the checkpoint wins. ## What the sink must provide The sink is where the guarantee is usually lost. **File sink — exactly-once.** It maintains a `_spark_metadata` directory alongside the output listing the files committed by each batch. A reader that goes through Spark honours that log, so files written by a batch that failed midway are ignored, and a replayed batch does not double-count. The catch is that the log is Spark's, not the filesystem's: an external tool that globs the directory will see orphaned files from failed attempts. **Kafka sink — at-least-once.** The built-in sink does not wrap its writes in a Kafka transaction, so a replayed batch republishes its records. Deduplicate downstream on a business key, or accept duplicates. **`foreachBatch` — whatever you build.** It hands you an ordinary batch DataFrame plus the `batchId`. The standard patterns are a `MERGE` keyed on business identity, an overwrite of a partition deterministically derived from the batch, or a small table recording the highest committed `batchId` that the write consults transactionally. Note that `foreachBatch` provides at-least-once by default — the function can be invoked more than once for the same `batchId`, and idempotency is entirely your responsibility. **`foreach`** operates per record with `open(partitionId, epochId)`/`process`/`close`, and the same reasoning applies at row granularity. ## Stateful queries State is checkpointed per batch alongside the offsets, so a replayed batch also rewinds state to the version that existed before it ran. That is why a stateful query's correctness after failure depends on the state store checkpoint being intact, and why restoring an old checkpoint against a source whose retention has already expired the corresponding offsets produces a query that cannot start. ## Things that quietly break it Sharing a checkpoint directory between two queries. Deleting the checkpoint to "reset" a query and then reprocessing into a non-idempotent sink. Performing side effects inside a `map` or UDF — those run per task attempt and are re-executed on retry with no batch-level protection. Writing to an external system from `foreachBatch` without using `batchId`. And assuming continuous processing mode carries the same guarantee: it is at-least-once and experimental. ## How to answer the question in an interview Name the three parties — replayable source, offset write-ahead log plus commit log in the checkpoint, idempotent or transactional sink — and then say which of your actual sinks satisfies the third. That last sentence is what separates a memorised answer from an operational one.