Definition
- Stateful stream processing – streaming computation that remembers information across events rather than treating each event independently.
- Keyed state – state scoped per key (e.g., per trading symbol), where the engine routes all events for one key to the same place.
- Windowed aggregation – computing over bounded time buckets of an unbounded stream (tumbling or sliding windows).
- State backend – where keyed state lives: in-memory for speed or RocksDB for large state exceeding RAM.
- Checkpointing – periodic consistent snapshots of all state plus source offsets for exactly-once fault tolerance.
- Time-to-live (TTL) – automatic eviction of stale keys to manage state size.
Imagine the National Bank of India's trading desk watching an endless stream of trades. Stateless processing would forget each trade instantly. Stateful processing is like a clerk keeping a running tally sheet: for each stock symbol, how much volume traded this minute. Keyed state is one tally line per symbol. The window is the minute boundary: at the end of each minute the clerk reads the tally, picks the top five busiest symbols, writes them on the board, and wipes the sheet for the next minute. Checkpointing is the clerk photographing the tally sheet every few seconds, so if they faint, a replacement resumes from the photo with no trade lost or double-counted.
Understanding the Concepts
What is the difference between stateless and stateful processing?
Stateless operations like a simple filter treat every event independently and need no memory. Stateful operations remember something across events: a running count, a running sum, the last-seen value, or a full aggregation. This is what makes aggregations, joins over time, and pattern detection possible on streams.
package com.bank.streaming;
import java.math.BigDecimal;
public class StatelessVsStateful {
public static void main(String[] args) {
// Stateless: each event independent
System.out.println("Stateless filter:");
System.out.println(" trade.amount > 10000 ? pass : drop");
System.out.println(" No memory needed.");
System.out.println();
// Stateful: remembers across events
System.out.println("Stateful aggregation:");
System.out.println(" Running sum of RELIANCE volume");
System.out.println(" Each trade updates the sum.");
System.out.println(" Requires keyed state per symbol.");
}
}
Stateless filter:
trade.amount > 10000 ? pass : drop
No memory needed.
Stateful aggregation:
Running sum of RELIANCE volume
Each trade updates the sum.
Requires keyed state per symbol.
What is keyed state?
Keyed state is state scoped per key. You partition the stream by a key (like trading symbol) and the engine maintains independent state for each key. All events for one key are routed to the same place so the state stays consistent.
package com.bank.streaming;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;
public class KeyedStateExample {
// Simulating per-symbol keyed state
static Map<String, BigDecimal> volumeBySymbol = new HashMap<>();
public static void main(String[] args) {
// Trade events stream in
processTrade("RELIANCE", new BigDecimal("50000"));
processTrade("TCS", new BigDecimal("30000"));
processTrade("RELIANCE", new BigDecimal("70000"));
processTrade("HDFC", new BigDecimal("45000"));
volumeBySymbol.forEach((sym, vol) ->
System.out.println(sym + " -> " + vol));
}
static void processTrade(String symbol, BigDecimal volume) {
volumeBySymbol.merge(symbol, volume, BigDecimal::add);
}
}
HDFC -> 45000
RELIANCE -> 120000
TCS -> 30000
What are tumbling and sliding windows?
A tumbling window is fixed and non-overlapping, like each one-minute interval. A sliding window overlaps, like a one-minute window advancing every ten seconds. Tumbling is for "every minute" queries; sliding is for "a rolling last minute updated frequently".
package com.bank.streaming;
public class WindowTypes {
public static void main(String[] args) {
System.out.println("Tumbling window (1 min):");
System.out.println(" [09:00-09:01) [09:01-09:02) ...");
System.out.println(" Non-overlapping. Each event in one window.");
System.out.println();
System.out.println("Sliding window (1 min, slide 10 sec):");
System.out.println(" [09:00-09:01) [09:00:10-09:01:10) ...");
System.out.println(" Overlapping. One event in multiple windows.");
System.out.println();
System.out.println("Top 5 symbols per minute -> tumbling.");
System.out.println("Rolling alert over last minute -> sliding.");
}
}
Tumbling window (1 min):
[09:00-09:01) [09:01-09:02) ...
Non-overlapping. Each event in one window.
Sliding window (1 min, slide 10 sec):
[09:00-09:01) [09:00:10-09:01:10) ...
Overlapping. One event in multiple windows.
Top 5 symbols per minute -> tumbling.
Rolling alert over last minute -> sliding.
What are state backends?
The state backend is where keyed state lives. An in-memory backend (HashMapStateBackend) is fast but bounded by RAM. A RocksDB backend (EmbeddedRocksDBStateBackend) spills to local disk so state can be far larger than memory, which matters when you key by millions of symbols or accounts.
package com.bank.streaming;
public class StateBackends {
public static void main(String[] args) {
System.out.println("In-memory (HashMapStateBackend):");
System.out.println(" Fast. Limited by JVM heap.");
System.out.println(" Good for: small state, few keys.");
System.out.println();
System.out.println("RocksDB (EmbeddedRocksDBStateBackend):");
System.out.println(" Spills to disk. Terabytes of state.");
System.out.println(" Good for: millions of keys.");
System.out.println(" Supports incremental checkpoints.");
}
}
In-memory (HashMapStateBackend):
Fast. Limited by JVM heap.
Good for: small state, few keys.
RocksDB (EmbeddedRocksDBStateBackend):
Spills to disk. Terabytes of state.
Good for: millions of keys.
Supports incremental checkpoints.
How does checkpointing provide fault tolerance?
Checkpointing periodically snapshots all keyed state plus source offsets as a consistent image. On failure the engine restores the latest checkpoint and replays from the saved offsets, giving exactly-once state: counts are neither lost nor doubled.
package com.bank.streaming;
public class CheckpointingExplained {
public static void main(String[] args) {
System.out.println("Checkpoint process:");
System.out.println(" 1. Barrier injected into stream");
System.out.println(" 2. Each operator snapshots its state");
System.out.println(" 3. Kafka offsets saved with state");
System.out.println(" 4. Snapshot stored to S3/HDFS");
System.out.println();
System.out.println("On failure:");
System.out.println(" 1. Restore latest checkpoint");
System.out.println(" 2. Reset Kafka to saved offsets");
System.out.println(" 3. Replay events from there");
System.out.println(" 4. State is exactly-once correct");
}
}
Checkpoint process:
1. Barrier injected into stream
2. Each operator snapshots its state
3. Kafka offsets saved with state
4. Snapshot stored to S3/HDFS
On failure:
1. Restore latest checkpoint
2. Reset Kafka to saved offsets
3. Replay events from there
4. State is exactly-once correct
How do you compute top 5 symbols by volume every minute?
The recipe: key the trade stream by symbol, sum each symbol's volume within a one-minute tumbling window, and when the watermark passes the end of the minute, sort descending and emit the top five. Then reset for the next minute.
# PySpark Structured Streaming: top 5 symbols per minute
from pyspark.sql.functions import window, col, sum as _sum
from pyspark.sql.functions import row_number
from pyspark.sql.window import Window
trades = spark.readStream.format("kafka") \
.option("subscribe", "trades").load()
per_symbol = trades.groupBy(
window(col("event_time"), "1 minute"),
col("symbol")
).agg(_sum("volume").alias("total_vol"))
w = Window.partitionBy("window") \
.orderBy(col("total_vol").desc())
top5 = per_symbol.withColumn("rank", row_number().over(w)) \
.filter(col("rank") <= 5)
top5.writeStream.outputMode("complete") \
.format("console").start()
What should a senior engineer know about state management?
Manage state size with time-to-live so stale keys are evicted. Choose tumbling vs sliding to match the business question. Use incremental checkpoints with RocksDB for large state. Ensure the ranking is deterministic.
package com.bank.streaming;
public class SeniorStateManagement {
public static void main(String[] args) {
System.out.println("Senior state concerns:");
System.out.println(" 1. TTL: evict stale keys (old symbols)");
System.out.println(" 2. Incremental checkpoints for large state");
System.out.println(" 3. Checkpoint interval vs recovery time");
System.out.println(" 4. RocksDB tuning for millions of keys");
System.out.println(" 5. Deterministic ranking (tie-breaking)");
}
}
Senior state concerns:
1. TTL: evict stale keys (old symbols)
2. Incremental checkpoints for large state
3. Checkpoint interval vs recovery time
4. RocksDB tuning for millions of keys
5. Deterministic ranking (tie-breaking)
What are common interview follow-up questions?
- What is the difference between a tumbling and a session window?
- How does Flink's checkpoint barrier protocol work?
- What happens if a key's state grows unbounded?
- How do you handle late data that arrives after a window closes?
- What is the difference between in-memory and RocksDB state backends?
Visual Overview
Top 5 Symbols by Volume Every Minute
─────────────────────────────────────
Trade Stream
┌──────────────────────────────────┐
│ RELIANCE 50k | TCS 30k | HDFC 45k │ ...
└──────────┬───────────────────────┘
│
│ keyBy(symbol)
▼
┌──────────────────────────────────┐
│ Per-Symbol Keyed State │
│ RELIANCE: 120k │
│ TCS: 30k │
│ HDFC: 45k │
│ ICICI: 80k │
│ SBI: 60k │
└──────────┬───────────────────────┘
│
│ Tumbling Window (1 min)
│ Watermark passes end
▼
┌──────────────────────────────────┐
│ Rank + Top 5 │
│ 1. RELIANCE 120k │
│ 2. ICICI 80k │
│ 3. SBI 60k │
│ 4. HDFC 45k │
│ 5. TCS 30k │
└──────────────────────────────────┘
Key Takeaway
Stateful stream processing is streaming computation that remembers information across events instead of treating each event independently, and it is what makes aggregations, time-based joins, and pattern detection possible on an unbounded stream. It rests on four ideas. Keyed state is state scoped per key, so you partition the stream by something like trading symbol and the engine keeps independent state for each key. Windowed aggregations bound the infinite stream into finite pieces, with tumbling windows for fixed non-overlapping intervals and sliding windows for rolling periods. The state backend determines where state lives: in memory for speed or RocksDB for large state. And checkpointing periodically snapshots all state plus source offsets for exactly-once fault tolerance.
For the specific task of the top five symbols by volume every minute, the recipe is a per-key aggregation inside a tumbling window followed by a ranking at window close. Key the trade stream by symbol, sum each symbol's volume within a one-minute tumbling window, and when the watermark passes the end of the minute, sort descending and emit the top five.
A senior engineer manages state size with time-to-live to evict stale keys, picks tumbling versus sliding to match the business question, uses incremental checkpoints with RocksDB for large state, and keeps the ranking deterministic.
In a banking context at the National Bank of India's trading analytics, an endless stream of trades was keyed by symbol, each symbol's volume accumulated in a one-minute tumbling window backed by RocksDB, and at every minute boundary the job emitted the five busiest symbols. Because state was checkpointed, even a mid-stream crash kept volumes correct without double-counting.
Remember one sentence: stateful streaming remembers per-key state inside time windows, snapshotted by checkpoints, enabling real-time aggregations like top-N rankings over an infinite stream.
---