Spark Structured Streaming state keeps growing: watermarks that don't work, unbounded dropDuplicates and stream-stream joins

#Symptoms

• Micro-batch duration grows steadily from seconds to minutes over days.
• Executors fail with out-of-memory errors or long GC pauses; restarts help briefly.
• The checkpoint folder's state/ directory keeps getting bigger.

Stateful operators (aggregations, dropDuplicates, stream-stream joins, flatMapGroupsWithState, transformWithState) keep state between batches. If nothing tells Spark when old state can be dropped, it keeps everything forever.

#Step 1: read the state metrics

Each batch reports state metrics in StreamingQueryProgress:

p = query.lastProgress
for op in p["stateOperators"]:
    print(op["operatorName"], op["numRowsTotal"], op["numRowsUpdated"],
          op["memoryUsedBytes"], op.get("numRowsDroppedByWatermark"),
          op.get("customMetrics", {}))
print(p["eventTime"])          # includes the current watermark

Log lastProgress to your metrics system with a StreamingQueryListener, so you can graph these over days.

#Step 2: fix the watermark

A watermark tells Spark how late data can be. State older than the watermark is removed.

from pyspark.sql import functions as F

events = (spark.readStream.format("kafka")...
          .select(F.from_json("value", schema).alias("e")).select("e.*"))

counts = (events
  .withWatermark("event_time", "30 minutes")              # BEFORE the aggregation
  .groupBy(F.window("event_time", "5 minutes"), "device_id")
  .count())

Common reasons the watermark doesn't clean state:

For multiple inputs, spark.sql.streaming.multipleWatermarkPolicy can be max, which moves faster but drops more late data from slower inputs.

#Step 3: deduplication without unbounded state

dropDuplicates(["event_id"]) on a stream remembers every event_id forever.

Options:

# Include the event-time column so state can expire (classic approach)
deduped = (events.withWatermark("event_time", "2 hours")
                 .dropDuplicates(["event_id", "event_time"]))

# Spark 3.5+: deduplicate within the watermark delay, keyed only on event_id
deduped = (events.withWatermark("event_time", "2 hours")
                 .dropDuplicatesWithinWatermark(["event_id"]))

dropDuplicatesWithinWatermark keeps each key only as long as the watermark delay, even when duplicates have slightly different timestamps.

#Step 4: stream-stream joins

Both sides need a watermark and the join condition needs a time-range constraint, or Spark must keep all rows of both streams:

impressions = imps.withWatermark("imp_time", "2 hours")
clicks = clks.withWatermark("click_time", "3 hours")

joined = impressions.join(
    clicks,
    F.expr("""
      clickAdId = impressionAdId AND
      click_time >= imp_time AND
      click_time <= imp_time + interval 1 hour
    """))

Outer joins also need watermarks to emit unmatched rows, and only emit them once the watermark passes.

#Step 5: use the RocksDB state store

The default HDFS-backed state store keeps state in executor JVM memory. Large state causes GC pressure. RocksDB keeps it on local disk with a memory cache:

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

Changelog checkpointing (Spark 3.4+) uploads only changes each batch instead of full snapshots. Changing the state store provider on an existing query requires a new checkpoint in most versions; plan it like a redeployment.

Also check spark.sql.shuffle.partitions: it's fixed at the first run of a query and sets the number of state partitions. Too few gives huge partitions; it can't be changed without a new checkpoint.

#Step 6: when you must reset state

Changing watermarks or keys on an existing checkpoint may fail with state schema errors, or be silently incompatible.

Caution

Starting from a new checkpoint loses state and the source offsets stored with it. Decide the restart point explicitly (Kafka startingOffsets, or a timestamp), expect duplicates or gaps around the switch, and make the sink idempotent.

Read existing state for debugging (Spark 3.5+/4.0 state data source; availability depends on version and platform):

state = spark.read.format("statestore").load("/checkpoints/device_counts")
state.groupBy().count().show()

#Prevent it next time

• Review every stateful operator for "how does state expire?" before deploying.
• Graph numRowsTotal and memoryUsedBytes per operator and alert on steady growth.
• Default to RocksDB with changelog checkpointing for any non-trivial state.
• Validate event-time quality (no future timestamps) before the watermark.