What are exactly-once semantics in Spark streaming?

Summary

• Exactly-once means each input record affects the output once, even after failures. In Structured Streaming, the Spark documentation ties end-to-end exactly-once to replayable sources (which have offsets) plus idempotent sinks.
• Checkpointing and write-ahead logs record progress, not data. The engine saves the offset range being processed in each trigger and the running aggregates, so a restarted query continues where it left off.
• The default micro-batch mode is exactly-once, while continuous processing is experimental and at-least-once. The older DStreams engine, Spark Streaming, is a legacy project.
• The sink decides the end result. The file sink is exactly-once, the Kafka and foreach sinks are at-least-once, and foreachBatch depends on your code. Delta Lake adds idempotent writes through txnAppId and txnVersion.

What are exactly-once semantics in Spark streaming?

Streaming systems describe their guarantees by how often each record is processed. At most once means a record is processed once or not at all. At least once means one or more times, so no data is lost but there may be duplicates. Exactly once means no data is lost and no data is processed multiple times. Spark's current engine is Structured Streaming. The older Spark Streaming (DStreams) engine is a legacy project with no further updates, so this page covers Structured Streaming.

How does Structured Streaming provide exactly-once?

The Spark documentation says that, using replayable sources and idempotent sinks, Structured Streaming can ensure end-to-end exactly-once semantics under any failure. Three pieces work together:

Piece	What it does	Why it matters
Replayable source	Every streaming source is assumed to have offsets, such as Kafka offsets, to track the read position	After a failure, the engine can read the same range again
Checkpoint and write-ahead log	The engine records the offset range being processed in each trigger and saves running aggregates to the checkpoint location	A restarted query continues where it left off
Idempotent sink	Sinks are designed to be idempotent for handling reprocessing	Replaying a batch does not duplicate output

The log holds offset ranges rather than the records, which is why the source must be able to replay the data.

How do I turn it on?

Set a checkpoint location as a writer option. It is an option on the DataStreamWriter, not a method, and it has to be a path in an HDFS compatible file system. The file sink then gives exactly-once output:

(df.writeStream
   .format("parquet")
   .option("checkpointLocation", "/path/to/checkpoint")
   .option("path", "/path/to/output")
   .start())

Which sinks keep the guarantee?

Sink	Fault-tolerant	Note
File sink	Yes (exactly-once)	Append mode
Kafka sink	Yes (at-least-once)	Kafka only supports at-least-once write semantics, so a unique key can help deduplicate when reading
Foreach sink	Yes (at-least-once)	Divide the write logic into open, process and close
ForeachBatch sink	Depends on the implementation	Use the batchId to deduplicate
Console and memory sinks	No	For debugging only

A Delta table sink adds its own protection: the transaction log lets Delta Lake guarantee exactly-once processing even with other streams or batch queries running against the table. For foreachBatch writes, Delta Lake 2.0.0 and above accepts the txnAppId and txnVersion writer options, so a re-run batch is recognized as a duplicate:

app_id = "orders-stream"  # a unique string for this application

def write_batch(batch_df, batch_id):
    (batch_df.write.format("delta")
        .option("txnAppId", app_id)
        .option("txnVersion", batch_id)
        .mode("append")
        .save("/path/to/delta/table"))

query = (df.writeStream
    .foreachBatch(write_batch)
    .option("checkpointLocation", "/path/to/checkpoint")
    .start())

What do continuous processing and Real-Time Mode change?

Continuous processing is an experimental mode with at-least-once fault-tolerance guarantees. The documentation lists only map-like operations, such as projections and selections, as supported, with no aggregations, and foreachBatch does not work with it. Spark 4.1.0 added Real-Time Mode as a separate low-latency mode, so check the documentation for your Spark version before assuming either set of guarantees applies to it.

How does Databricks handle checkpoints?

• Checkpoint contents: the Databricks documentation lists offsets (the source offsets processed in each micro-batch), commits (which micro-batches were committed to the sink), state and metadata. Each query needs its own checkpoint location.
• Lakeflow Pipelines: checkpoint locations are internal to the pipeline and not accessible to users, so you typically don't manage them. The open-source Spark Declarative Pipelines framework also handles checkpoints for you.
• Your own streams: specify checkpointLocation, for example a Unity Catalog volume path such as /Volumes/catalog/schema/volume/path.

FAQs

What happens if I delete the checkpoint?

The next run of the query begins fresh. For a Kafka source, startingOffsets applies when a new query starts, and its default for streaming queries is "latest", so records produced while the query was stopped are not read unless you set earlier offsets. A Delta foreachBatch writer also needs a different txnAppId, because the new batch IDs start from 0 and writes with the old ID would be ignored.

Can I get exactly-once with continuous processing?

No. The documentation gives at-least-once guarantees for continuous mode. Run the query in micro-batch mode to get exactly-once processing.