What is a state store?

Summary

• A state store is the place a streaming query keeps its state between micro-batches. The Spark docs call it "a versioned key-value store which provides both read and write operations", and Structured Streaming uses a state store provider to handle stateful operations across batches.
• Apache Spark ships two built-in providers. The HDFS state store provider is the default and keeps state in an in-memory map backed by files in an HDFS-compatible file system. The RocksDB provider, added in Spark 3.2, keeps state in native memory and on local disk.
• RocksDB is opt-in in Apache Spark. Set spark.sql.streaming.stateStore.providerClass to org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider. On Databricks Runtime 17.3 and above, RocksDB is the default.
• State grows unless the query can forget it. Watermarks let the engine clean up old state, and transformWithState state variables can carry a TTL.
• State is checkpointed and restored after a restart, so changes to a stateful operation's schema are not allowed between restarts.

What is a state store?

A state store lets a streaming query remember things between micro-batches. A running count, a set of seen keys and the buffered side of a join all live there. It is a versioned key-value store, and updates are applied in transactional sets that each increment the store's version. The state lives with the executors, and Structured Streaming "automatically checkpoints the state data to fault-tolerant storage" and restores it after a restart.

Which operations use a state store?

The Spark docs name streaming aggregation, streaming dropDuplicates, stream-stream joins, mapGroupsWithState and flatMapGroupsWithState. Since Spark 4.0 the docs encourage transformWithState for new custom stateful applications, and its StatefulProcessorHandle provides methods to interact with the underlying state store.

Which state store providers does Spark have?

Provider	Where state lives	How it is made durable	Why you would pick it
HDFS state store provider (default)	An in-memory map in the executor JVM	Backed by files in an HDFS-compatible file system	It is the default implementation
RocksDB state store provider (Spark 3.2 and later)	Native memory and local disk	Changes are saved to the checkpoint location, with the same fault-tolerance guarantees	You keep millions of keys and see long JVM garbage collection pauses

You can also write your own provider by extending the StateStoreProvider interface.

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider",
)

How does a state store survive failures?

For the HDFS provider, updates are applied in transactional sets and each set increments the store's version. Spark can then re-execute updates on the correct version after a retry. RocksDB changes are saved to the checkpoint location. Changelog checkpointing, controlled by spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled and off by default, uploads only the changes since the last checkpoint instead of snapshots. The Spark 4.2.0 release notes also list row checksums for corruption detection and automatic snapshot repair for the state store.

How do you keep state from growing without bound?

Watermarks tell the engine when old state can go. They let it "automatically track the current event time in the data and attempt to clean up old state accordingly". For aggregations the docs list conditions: Append or Update output mode, an event-time column or window in the aggregation, and withWatermark called on that column before the aggregation. Without a watermark, dropDuplicates "stores the data from all the past records as state", and an inner stream-stream join's state grows indefinitely until you add conditions.

# Without watermark using guid column
streamingDf.dropDuplicates(["guid"])

# With watermark using guid and eventTime columns
streamingDf \
  .withWatermark("eventTime", "10 seconds") \
  .dropDuplicates(["guid", "eventTime"])

In transformWithState, a state variable can take an optional TTL that evicts it after the given duration.

How do you monitor state?

The Spark Web UI has a Structured Streaming tab for micro-batch queries. It shows "Aggregated Number Of Total State Rows", "Aggregated Number Of Updated State Rows", "Aggregated State Memory Used In Bytes" and "Aggregated Number Of State Rows Dropped By Watermark". The progress JSON from query.lastProgress carries a stateOperators array. With the RocksDB provider, spark.sql.streaming.stateStore.rocksdb.trackTotalNumberOfRows defaults to true, and turning it off makes the total row count report as 0.

How does Databricks approach state stores?

"RocksDB is the default state store provider in Databricks Runtime 17.3 and above." On earlier versions, set spark.sql.streaming.stateStore.providerClass to com.databricks.sql.streaming.state.RocksDBStateStoreProvider. Changelog checkpointing is on by default from 17.3, and Databricks recommends it for all stateful queries. From 17.3, Databricks caps RocksDB memory per node automatically; on earlier versions you set spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage and spark.sql.streaming.stateStore.rocksdb.maxMemoryUsageMB. The state management scheme can't change between restarts, so a query started with the default scheme needs a new checkpoint location to switch.

FAQs

Is RocksDB the default state store in Apache Spark?

No. The HDFS state store provider is the default, and RocksDB is opt-in through spark.sql.streaming.stateStore.providerClass.

Can I use an external state store?

Yes, by extending the StateStoreProvider interface with your own provider. Spark's two built-in providers both persist state to the checkpoint location.

Can I change a stateful query between restarts?

Not its state schema. The docs say additions, deletions or schema modifications to stateful operations "are not allowed between restarts". That covers aggregation keys and aggregates, deduplication columns, and the schema, equi-join columns and join type of stream-stream joins.