How does Spark Streaming handle stateful processing?

Stateful processing means a streaming query remembers information across micro-batches, such as running counts, open sessions or the rows waiting for a join partner. In Structured Streaming, Spark keeps this state in a fault-tolerant state store on the executors, versioned and checkpointed with each micro-batch, and watermarks decide when old state can be removed.

Built-in stateful operations

• Aggregations: groupBy(...).count(), running totals, and event-time windows (tumbling, sliding and session windows with session_window).
• Stream–stream joins: each side's rows are kept until a matching row can no longer arrive.
• Deduplication: dropDuplicates or dropDuplicatesWithinWatermark keeps the keys already seen.

from pyspark.sql import functions as F

sessions = (events
    .withWatermark("event_time", "30 minutes")
    .groupBy(F.session_window("event_time", "10 minutes"), "user_id")
    .agg(F.count("*").alias("events")))

Custom state For logic the built-in operators can't express, such as alerting when a device's reading stays high for 3 readings in a row, use arbitrary stateful processing:

• transformWithState (Spark 4.0 and later; transformWithStateInPandas in Python) is the current API. It supports several state variables per key (value, list and map state), time-to-live for state, processing-time and event-time timers, and schema evolution of state.
• applyInPandasWithState (Python) and mapGroupsWithState / flatMapGroupsWithState (Scala, Java) are the older APIs, still supported.

Where the state lives

• The default state store keeps state in executor memory and writes versioned files to the checkpoint location.
• The RocksDB state store (spark.sql.streaming.stateStore.providerClass set to the RocksDB provider) keeps state on local disk with a memory cache, which handles much larger state with less garbage collection; it is the usual choice for production.
• Changelog checkpointing for RocksDB writes only the changes in each batch, which shortens checkpoint time.

Fault tolerance Each micro-batch's offsets and state version are written to the checkpoint. After a failure, the query restarts from the last committed batch, reloads the matching state version and reprocesses from the source, giving exactly-once results with replayable sources and idempotent sinks. The checkpoint must stay on reliable storage, and some changes to the query (such as changing grouping keys) are incompatible with existing state.

Keeping state under control

• Always use watermarks (or state time-to-live in transformWithState) so state for finished keys is removed.
• Watch state metrics in the query progress and the Structured Streaming UI: number of state rows and memory used.
• Partition state sensibly: the number of shuffle partitions is fixed for a query once it has a checkpoint.

The old DStream approach DStreams used updateStateByKey and mapWithState. They are legacy; new stateful jobs use Structured Streaming.