Why architecture matters here
Stateful streaming matters because the valuable stream-processing operations are stateful, and managing state over unbounded streams correctly is genuinely hard. Aggregations (counts, sums per window or key), joins (correlating two streams), deduplication (removing duplicates), and sessionization — the operations that turn a raw stream into insights — all require state (accumulating and remembering across records). But over an unbounded stream, state management faces three hard problems that stateless processing doesn't: unbounded growth (state accumulates forever unless bounded — memory exhaustion), durability (state must survive failures without loss or duplication — exactly-once), and time and lateness (streams have out-of-order and late data — the state and results must handle it correctly). Spark Structured Streaming provides a coherent solution to all three (watermarks bound growth, checkpointing provides durability, watermarks and output modes handle lateness) — making stateful streaming tractable. For any streaming workload beyond stateless transformation (which is most valuable streaming), understanding this state management is essential, because getting it wrong (unbounded state, lost state on failure, incorrect late-data handling) is where streaming systems fail.
The watermark mechanism is the architectural key to bounding state, and it's the crucial concept for sustainable stateful streaming. A watermark is a threshold in event time — 'I don't expect data older than (max event time seen − delay threshold)'. It does two critical things. First, it bounds state: for windowed aggregations, once the watermark passes a window's end (plus the allowed lateness), no more data for that window is expected, so the window's state can be finalized and evicted (cleaned up) — this is what prevents unbounded state growth (old windows' state is removed once the watermark says they're complete). Without watermarks, aggregation state would grow forever (every window kept indefinitely, waiting for possible late data). Second, it defines late-data handling: data arriving after the watermark passed its window is 'too late' (dropped, since the window's already finalized) — the watermark is the line between 'late but accepted' and 'too late, dropped'. This is a fundamental tradeoff: a larger watermark delay (more allowed lateness) accepts more late data (better completeness) but keeps state longer (more state, more memory); a smaller delay bounds state tighter (less memory) but drops more late data (less completeness). Tuning the watermark — balancing completeness against state size — is the central stateful-streaming decision, and understanding that watermarks make unbounded-stream state management possible (by defining when state is complete and evictable) is understanding the core of stateful streaming.
And the checkpointing-for-exactly-once design is what makes stateful streaming reliable despite failures. Streaming is long-running (a stream runs indefinitely), so failures (instance crashes, restarts) are inevitable — and stateful processing must recover without losing or duplicating (exactly-once). Checkpointing provides this: Spark periodically checkpoints the state (durably persisting the state stores) and the stream progress (the offsets processed) together — so on failure, processing recovers from the checkpoint (restoring the state and resuming from the recorded offset), reprocessing exactly the data since the checkpoint without loss (the state is restored) or duplication (the offsets prevent reprocessing already-processed data into the state twice). This exactly-once state recovery is critical for stateful correctness (an aggregation must not lose counts or double-count across a failure) and is a managed feature (Spark handles the checkpointing and recovery) — but it introduces operational concerns (checkpoint storage, checkpoint frequency affecting recovery time and overhead, checkpoint compatibility across code changes). Understanding that checkpointing provides the durable, exactly-once state recovery that makes long-running stateful streaming reliable — and its operational implications — is essential to operating stateful streaming.
The architecture: every piece explained
Top row: the state foundation. The state store holds keyed state, partitioned (state for each key, distributed across partitions) — the accumulating state for aggregations, joins, dedup. Checkpointing durably persists the state stores and the stream offsets together — enabling exactly-once recovery (restore state, resume from offset). Watermarks are event-time bounds — 'no data older than (max seen − delay) expected' — bounding state (windows past the watermark are complete and evictable) and defining late-data handling (data past the watermark is too late). Output modes control result emission: append (only new final results — for windowed aggregations after the watermark finalizes them), update (changed results — for incrementally-updated aggregations), complete (the full result table each trigger — for aggregations where the whole result is re-emitted) — matching the operation and sink.
Middle row: the stateful operations. Windowed aggregation: aggregating over time windows (tumbling, sliding) — state per window, finalized and evicted as the watermark passes. Stream-stream joins: joining two streams — buffering records from each (state) to match against the other, bounded by watermarks (how long to buffer waiting for matches — beyond the watermark, unmatched records are dropped and their state evicted). Dedup and arbitrary state: deduplication (remembering seen keys to drop duplicates, bounded by watermark) and arbitrary stateful processing (flatMapGroupsWithState/mapGroupsWithState — custom state logic with explicit state management and timeout-based cleanup) — the flexible stateful operations. State cleanup: watermark-driven eviction — state for completed windows/expired joins/timed-out keys is evicted as the watermark advances, bounding state.
Bottom rows: backend and correctness. RocksDB state backend: for large state, the RocksDB backend stores state off-heap (in RocksDB, on local disk) rather than in the JVM heap — avoiding heap pressure and GC issues for big state (large aggregations, many keys), enabling much larger state than the default heap-based backend. Late data and recovery: late data (within the watermark) is incorporated (updating the relevant window/state), too-late data (past the watermark) is dropped; recovery (from checkpoints) restores state exactly-once after failures — the correctness under lateness and failure. The ops strip: state size (monitoring and bounding state — the watermark bounds it, RocksDB handles large state, but unbounded operations or too-large watermarks grow state; managing state size is central), watermark tuning (balancing completeness against state size — the delay threshold tradeoff), and checkpoint management (checkpoint storage, frequency, and compatibility across code changes — checkpoints tie to the query structure, so code changes can break checkpoint compatibility).
End-to-end flow
Trace a windowed aggregation with watermarks. A stream of events is aggregated into 1-minute tumbling windows (count per minute), with a watermark of 5 minutes ('accept data up to 5 minutes late'). As events arrive, they're aggregated into their event-time windows (state per window, in the state store). The watermark advances with the max event time seen. When the watermark passes a window's end plus the 5-minute allowed lateness (e.g., the 10:00-10:01 window is complete when the watermark reaches 10:06), that window's result is finalized (emitted in append mode) and its state evicted (cleaned up) — bounding the state (old windows removed once complete). Late data: an event with event time 10:00:30 arriving at 10:04 (within the 5-minute watermark) is incorporated into the 10:00-10:01 window (updating its count — correct late-data handling); the same event arriving at 10:07 (past the watermark, the window already finalized) is too late and dropped. The watermark bounded the state (windows evicted once complete) and handled lateness (late-but-within-watermark incorporated, too-late dropped) — the core stateful-streaming mechanism.
The state-size and RocksDB vignettes show the scaling challenge. A high-cardinality aggregation (aggregating per user, millions of users, over long windows) accumulates large state (millions of keys' state). With the default heap-based state backend, this large state pressures the JVM heap (GC issues, memory limits). The team switches to the RocksDB state backend (state off-heap in RocksDB on local disk) — handling the large state without heap pressure, enabling much larger state than heap-based. A state-growth issue: a stream-stream join with a too-large watermark (buffering records for a long time waiting for matches) grows large join state (many buffered records); tightening the watermark (bounding how long to buffer) reduces the state (at the cost of dropping more unmatched late records) — the watermark-vs-state tradeoff managing join state size.
The recovery and checkpoint vignettes complete it. A recovery case: the streaming job's instance crashes; on restart, Spark recovers from the last checkpoint — restoring the state stores (the accumulated aggregation state) and resuming from the checkpointed offset (reprocessing exactly the data since the checkpoint) — exactly-once recovery, the aggregation correct across the failure (no lost or double-counted data). A checkpoint-compatibility case: the team changes the query (adding a field to the aggregation) and finds the new code can't recover from the old checkpoint (the checkpoint's state structure doesn't match the new query) — a checkpoint-compatibility break. They handle it carefully (some changes are compatible, others require a new checkpoint / state migration) — the operational reality that checkpoints tie to query structure. The consolidated discipline the team documents: manage state with watermarks (bounding state, handling lateness — the central mechanism), tune watermarks balancing completeness against state size, use the RocksDB backend for large state (off-heap, avoiding GC pressure), rely on checkpointing for exactly-once recovery (durable state and offsets), choose output modes matching the operation, monitor and bound state size, and manage checkpoint compatibility across code changes — because stateful streaming turns unbounded streams into aggregations, joins, and dedup, but managing state over unbounded data (bounding growth with watermarks, ensuring durability with checkpoints, handling lateness) is the hard part where correctness and scalability are won or lost.