Why is my state store growing without bound?

Summary

• State grows without bound when Spark has no rule for dropping it: no usable watermark, or custom state with no TTL and no removal.
• Call withWatermark on the event-time column before an aggregation or deduplication, and use Append or Update output mode; Complete mode keeps all aggregate state.
• Stream-stream joins also need an event-time join condition so old rows can leave state.
• In transformWithState (Apache Spark 4.0 and later), give each state variable a TTL (ttlDurationMs in PySpark) when you create it, or call clear().

Why is my state store growing without bound?

Your state store grows without bound because Spark has no rule that tells it when old state can go. Structured Streaming keeps state for aggregations, deduplication, stream-stream joins and arbitrary stateful operators, and removes entries only when a watermark, a TTL or your own code marks them finished. Declare a watermark Spark can use, bound joins with an event-time condition, and give custom state a TTL or clear it.

What stops state from being cleaned up?

Symptom	Likely cause	Fix
Windowed aggregation state keeps growing	No watermark, or withWatermark is on another column or after the aggregation	Call withWatermark on the aggregation's event-time column, before the aggregation
Aggregation state grows despite a watermark	Complete output mode "requires all aggregate data to be preserved"	Use Append or Update output mode
dropDuplicates state keeps growing	Without a watermark, "the query stores the data from all the past records as state"	Add a watermark and deduplicate on the ID plus event time, or use dropDuplicatesWithinWatermark
Stream-stream join state keeps growing	Both sides buffer past input to match future rows	Watermark both inputs and add a time-range or window join condition
transformWithState state keeps growing	No TTL, so state never expires	Pass ttlDurationMs, or call clear() when a key is done
Legacy flatMapGroupsWithState state keeps growing	Keys never time out or get removed	Set a GroupStateTimeout and call remove() for finished keys
Long GC pauses as state grows	The default HDFS-backed provider keeps state in executor JVM memory	Use the RocksDB provider; it moves state but doesn't delete it

How do watermarks bound state?

withWatermark(eventTime, delayThreshold) marks "a point in time before which we assume no more late data is going to arrive." Spark uses it "to minimize the amount of state that we need to keep for on-going aggregations." The docs list four conditions for cleaning aggregation state: Append or Update output mode, an aggregation on the event-time column or a window over it, a watermark on that same column, and withWatermark called before the aggregation.

from pyspark.sql.functions import window

windowedCounts = events \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(window(events.timestamp, "10 minutes", "5 minutes"), events.word) \
    .count()

query = windowedCounts.writeStream.outputMode("update").format("console").start()

Stream-stream joins need watermarks on both inputs plus an event-time constraint, such as clickTime <= impressionTime + interval 1 hour or a join on event-time windows. Without such conditions, the guide warns, "the size of streaming state will keep growing indefinitely."

How do I expire state in transformWithState?

transformWithState is the arbitrary stateful operator added in Apache Spark 4.0, "the next generation replacement" for mapGroupsWithState, flatMapGroupsWithState and applyInPandasWithState. PySpark added transformWithStateInPandas in 4.0.0 and the Row-based transformWithState in 4.1.0.
Set TTL per state variable in init: getValueState, getListState and getMapState each take an optional ttlDurationMs. Expired values "will be eventually removed from the state store," each update resets the expiry to current processing time plus the TTL, and "if ttl is not specified the state will never expire."

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor

class LastSeen(StatefulProcessor):
    def init(self, handle) -> None:
        # Each key's value expires one hour after its last update
        self.last_seen = handle.getValueState(
            "last_seen", "value TIMESTAMP", ttlDurationMs=60 * 60 * 1000)

    def handleInputRows(self, key, rows, timerValues):
        latest = max(map(lambda x: x.timestamp, rows))
        self.last_seen.update((latest,))
        yield Row(id=key[0], last_seen=latest)

    def close(self) -> None:
        pass

Run it with transformWithState (PySpark 4.1.0 and later) and timeMode="ProcessingTime"; other time modes reject TTL with STATEFUL_PROCESSOR_INCORRECT_TIME_MODE_TO_ASSIGN_TTL. The operator also needs spark.sql.streaming.stateStore.providerClass set to org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider. If you drop a state variable, call deleteIfExists in init so the engine purges its data.

How can I check what's in state?

• The Structured Streaming tab of the Spark UI charts "Aggregated Number Of Total State Rows" and "Aggregated State Memory Used In Bytes". Steady growth means old state isn't leaving.
• query.lastProgress reports stateOperators with numRowsTotal and numRowsUpdated.
• The experimental state data source (Spark 4.0 and later) reads state in a batch query: spark.read.format("statestore").load("<checkpointLocation>") returns key, value and partition_id, showing which keys linger.

How Databricks approaches state management

On Databricks, RocksDB is the default state store provider in Databricks Runtime 17.3 and above, with changelog checkpointing on by default, which Databricks recommends for all stateful queries. transformWithState is available in Databricks Runtime 16.2 and above; without TTL, the docs say, "you must handle state eviction to avoid out-of-memory errors." For Lakeflow Pipelines, the docs say to use watermarks for stateful aggregations, joins and deduplication (withWatermark in Python, WATERMARK ... DELAY OF INTERVAL in SQL).

FAQs

What watermark delay should I choose?

The lateness you're willing to wait for. Data within the threshold is aggregated, and later data "will start getting dropped," though the guarantee "is strict only in one direction." A window's state stays until the max event time seen minus the threshold passes the window's end, so a longer delay keeps state longer.

Does transformWithState TTL follow event time?

No. TTL counts from processing time and resets on each update. For event-time expiry, use EventTime mode, where the watermark "is used to expire timers": register a timer with registerTimer and call clear() in handleExpiredTimer.

Will switching to RocksDB stop the growth?

No. RocksDB keeps state "in the native memory and the local disk" instead of the JVM heap, which helps with GC pauses, but old entries still need a watermark, a TTL or explicit removal.