Structured Streaming internals: the checkpoint, the state store, and what a restart cannot change

• What is actually hard about streaming?
• What you should know already
• What does one micro-batch do?
• What does the checkpoint directory look like?
• Why can’t you change shuffle partitions after starting?
• How is state stored and compacted?
• How does the watermark actually move?
• Is state actually being evicted?
• What happens on restart?
• What should you watch in production?
• Common misconceptions
• A model worth keeping
• References
• Trademarks

TL;DR

• A micro-batch is bracketed by two writes: an entry in offsets/ before it runs, and an entry in commits/ after. Recovery is deciding which of those two a batch reached.
• The offset log records your configuration, not just your offsets. spark.sql.shuffle.partitions is written into every offset file, which is why a stateful query cannot change it across a restart.
• State lives in one directory per shuffle partition, written as one .delta per batch, compacted into a .snapshot every spark.sql.streaming.stateStore.minDeltasForSnapshot batches, default 10.
• The watermark is checkpointed twice on purpose: batchWatermarkMs in the offset file is what the batch used, nextBatchWatermarkMs in the commit file is what the next one will use.
• Watermark-driven eviction is visible in the progress metrics. A steady-state windowed aggregation held 40 state rows while updating 20 and removing 10 per batch, so state stopped growing without anyone configuring a retention.

What is actually hard about streaming?

Restarting a batch job is free. It reads its input, writes its output, and if it died halfway you run it again. A streaming query has no such luxury: it has consumed input that will not be replayed by default, it has emitted output that downstream systems have already seen, and it holds aggregates in memory that took hours to accumulate. A restart has to resume, not repeat.

Structured Streaming solves this with a loop that is unusually boring, and the boringness is the design. Before a micro-batch runs, the engine writes down exactly which input range it is about to process. After the batch finishes, it writes down that the batch completed. Everything else — exactly-once output, watermarks surviving restarts, state recovery — follows from those two files and the ordering between them.

This post opens that loop. Everything below was run on Spark 4.1.3 against a rate source with a windowed aggregation, and the files quoted are the ones that query actually wrote.

What you should know already

You should have written a Structured Streaming query and know what a trigger, a source and a sink are. Familiarity with event time versus processing time helps, though the distinction is re-established below where it matters.

Three terms are used precisely. A micro-batch is one iteration of the loop, identified by a batchId that only increases. State is the data a stateful operator keeps between batches, such as the running count for each open window. The watermark is the engine’s claim about how far event time has advanced, used to decide when state can be dropped.

The query these examples come from:

src = (spark.readStream.format("rate")
       .option("rowsPerSecond", 500).option("numPartitions", 2).load())

agg = (src.withWatermark("timestamp", "10 seconds")
       .groupBy(F.window(F.col("timestamp"), "5 seconds"), (F.col("value") % 10).alias("k"))
       .agg(F.count("*").alias("n")))

q = (agg.writeStream.outputMode("update").format("memory").queryName("agg")
     .option("checkpointLocation", "/work/ckpt")
     .trigger(processingTime="5 seconds").start())

What does one micro-batch do?

The loop is the same for every source and sink.

flowchart TB
    T[Trigger fires] --> P["Ask sources for the latest available offset"]
    P --> W1["Write offsets/&lt;batchId&gt;<br/>range + watermark + conf"]
    W1 --> PLAN["Plan the batch<br/>state store reads bound to this batchId"]
    PLAN --> RUN[Execute the micro-batch]
    RUN --> SINK[Write to the sink]
    SINK --> ST["Commit state: partition N writes N.delta"]
    ST --> W2["Write commits/&lt;batchId&gt;<br/>next watermark"]
    W2 --> T

The write-ahead order is the whole trick. offsets/<batchId> is durable before any processing starts, so a crash mid-batch leaves an offset entry with no matching commit entry. On restart the engine sees that asymmetry and re-runs that exact batch over that exact input range, which is what makes the recomputation deterministic rather than “whatever has arrived since”.

A batch that finished leaves both files, and the engine moves on to the next id.

What does the checkpoint directory look like?

After ten batches of the query above, the checkpoint contained:

ckpt/
├── metadata                  # {"id":"515b32bd-e7e4-4b83-b2a2-c679d6064108"}
├── offsets/
│   ├── 0 ... 9               # one per batch, written BEFORE the batch runs
├── commits/
│   ├── 0 ... 9               # one per batch, written AFTER it succeeds
└── state/
    └── 0/                    # stateful operator id
        ├── _metadata/schema  # key and value schema for this operator
        ├── 0/                # one directory per shuffle partition
        │   ├── 1.delta ... 10.delta
        ├── 1/
        ├── 2/
        └── 3/

metadata holds the query id, which is what makes a checkpoint belong to a query rather than to a directory path.

The number of state partition directories is spark.sql.shuffle.partitions. The query set it to 4, and there are exactly four. This is the fact that the next section turns into a constraint.

Why can’t you change shuffle partitions after starting?

Here is the offset file for batch 9, in full:

v1
{"batchWatermarkMs":1790079809515,"batchTimestampMs":1790079825001,"conf":{
 "spark.sql.streaming.stateStore.providerClass":"org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider",
 "spark.sql.streaming.stateStore.encodingFormat":"unsaferow",
 "spark.sql.streaming.multipleWatermarkPolicy":"min",
 "spark.sql.streaming.aggregation.stateFormatVersion":"2",
 "spark.sql.shuffle.partitions":"4",
 "spark.sql.streaming.join.stateFormatVersion":"2",
 "spark.sql.streaming.stateStore.compression.codec":"lz4"}}
44

Three things are stored, and only one of them is what the name suggests. The last line, 44, is the source offset. batchWatermarkMs is the watermark this batch ran with. And conf is a snapshot of the settings that determine how state is laid out and encoded.

That conf block is why several settings are effectively permanent. spark.sql.shuffle.partitions decides how many state partition directories exist and which key lands in which one. Changing it on restart would send a key to a different partition than the one holding its state, so the engine pins the original value instead. The same logic covers the state format versions and the encoding format: they describe how the bytes on disk are written, and the bytes on disk are already written.

The practical rule: choose the shuffle partition count before the first run of a stateful query, and treat it as part of the schema. Changing it later means a new checkpoint, which means rebuilding state.

The state schema file records its own constraint. For this query it holds:

{"type":"struct","fields":[
  {"name":"window","type":{"type":"struct","fields":[
     {"name":"start","type":"timestamp"},{"name":"end","type":"timestamp"}]},
   "metadata":{"spark.timeWindow":true,"spark.watermarkDelayMs":10000}},
  {"name":"_groupingexpression","type":"long"}]}

The key is the window plus the grouping expression, and the watermark delay is part of the recorded schema rather than a runtime setting. This is why changing a window duration or a watermark delay on an existing checkpoint is rejected rather than silently applied.

How is state stored and compacted?

Each partition directory accumulates one file per batch:

ckpt/state/0/0/
├── 1.delta
├── 2.delta
...
└── 10.delta

A .delta holds the changes that batch made to that partition’s state, not the whole state. Recovering a partition means replaying its deltas in order, which is cheap for ten files and expensive for ten thousand.

That is what snapshots are for. Every spark.sql.streaming.stateStore.minDeltasForSnapshot batches, default 10, a background maintenance task writes a .snapshot containing the full state, and recovery then starts from the snapshot and replays only the deltas after it. Old files are kept for spark.sql.streaming.minBatchesToRetain batches, default 100, so that a query can be restarted at an earlier batch.

The defaults this query ran with, read from the running engine:

HDFSBackedStateStoreProvider keeps the full state of each partition in executor memory, backed by those files. That is the property that decides whether your query scales: state size is bounded by executor heap. The RocksDB provider exists for state too large for that, trading heap for local disk and its own compaction.

Ten batches had run, and no .snapshot existed yet, which is consistent with a threshold of 10 and a maintenance task that runs on its own interval rather than inline with the batch.

How does the watermark actually move?

The watermark appears in two files, and they hold different values on purpose.

The offset file for batch 9 records batchWatermarkMs — the watermark that batch used when deciding what was late. The commit file for the same batch records what the next batch will use:

v1
{"nextBatchWatermarkMs":1790079814515}

The separation exists because a watermark can only be computed from data the engine has already seen. Batch 9 observes the maximum event time in its input, subtracts the configured delay, and that becomes batch 10’s threshold. Writing it into batch 9’s commit is what lets a restart at batch 10 begin with the correct threshold instead of starting over at zero.

Across three consecutive batches the watermark advanced by exactly the trigger interval:

2500 rows per batch is 500 rows per second across a five second trigger, and the watermark moves five seconds per batch because the rate source’s event time advances with wall clock. In a real stream the watermark moves with the data, not the clock, which is why a source that goes quiet freezes the watermark and stops state from being evicted.

Is state actually being evicted?

This is the question worth asking of any stateful query, and the progress metrics answer it directly. For the same three batches:

numRowsTotal holding steady while numRowsRemoved is non-zero is what a healthy stateful query looks like. Twenty rows updated and ten removed per batch, with the total flat at forty: windows are closing as fast as new ones open, and memory is not growing.

The failure mode is the same table with numRowsRemoved at zero and numRowsTotal climbing. That means the watermark is not advancing past any window’s end, and state will grow until the executor dies. The usual causes are a watermark delay larger than anyone intended, a source with no data so the watermark is frozen, or an aggregation with no watermark at all, where state is kept forever by design.

Note numShufflePartitions is 4, matching the state directories. The metric exists precisely because that number is pinned.

What happens on restart?

Recovery reads three things, in order.

The last offset file without a matching commit identifies an incomplete batch. That batch is re-run over the exact input range recorded, using the watermark recorded with it. If every offset file has a commit, the query starts the next batch id instead.

The state store is loaded per partition from the newest snapshot plus the deltas after it.

The conf block is applied, overriding whatever is in your submit command for the pinned settings.

Exactly-once output depends on the sink cooperating with this. A sink that is idempotent for a given batchId, which the file sink achieves through its own commit log, produces the same result whether a batch ran once or twice. A sink without that property — an arbitrary foreachBatch writing to an external system — gives at-least-once, and the duplicate is the re-run batch. The engine guarantees the recomputation is identical; it cannot guarantee your sink noticed.

What should you watch in production?

Four numbers from StreamingQueryProgress cover most failures.

numRowsTotal trending up across hours is unbounded state, and it is the single most common way a streaming job dies days after deployment.

Batch duration approaching the trigger interval means the query is about to fall behind. Once processing takes longer than the trigger, batches queue and latency grows without bound.

numInputRows collapsing to zero on a source that should have data is an upstream problem, and it also freezes the watermark, so it quietly stops state eviction at the same time.

The gap between batchTimestampMs and the watermark is how far behind event time is running. A growing gap means late data or a stalled partition.

Turn on spark.sql.streaming.metricsEnabled, which defaults to false, to get these through the metrics system rather than only in the progress object.

Common misconceptions

“Structured Streaming processes records one at a time.” It runs micro-batches. Continuous processing exists as a separate, limited execution mode; the default and the one described here is a loop of small batches.

“The checkpoint stores my data.” It stores offsets, watermarks, configuration and state. The input data is not in there, which is why a source that cannot replay a range breaks recovery.

“I can change any setting and restart.” Settings inside the offset log’s conf block are pinned, including the shuffle partition count. Changing them requires a new checkpoint.

“A watermark deletes late data.” It decides when state can be dropped. Rows later than the watermark are dropped from stateful operators because their window is gone, but the watermark is not a filter you can rely on for correctness of non-stateful paths.

“No data means nothing happens.” With spark.sql.streaming.noDataMicroBatches.enabled at its default of true, the engine still runs batches with no input, specifically so watermarks can advance and windows can close.

A model worth keeping

A streaming query is a loop that writes its intent, does the work, then writes its completion. offsets/ is the intent, commits/ is the completion, and the difference between them after a crash is the recovery plan.

State is a set of per-partition directories written as deltas and folded into snapshots, sized by whatever your executors can hold, and pruned only as fast as the watermark advances. Most production failures are one of two things: the watermark stopped moving, so state grew forever; or someone changed a setting the offset log had already pinned.

References

• Structured Streaming programming guide for the semantics these files implement
• Structured Streaming configuration for the spark.sql.streaming.* settings and defaults
• HDFSBackedStateStoreProvider for the delta and snapshot handling described above
• Logging a Spark stream for getting diagnostics out of a query you cannot restart
• Spark shuffle internals for what the shuffle partition count pinned in the offset log controls

Trademarks

Apache Spark, Apache and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.