Stateful Stream Processing
Counting, joining, windowing and deduplicating all require remembering something between records — which turns a job into a database you have to operate.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
Which streaming operations cannot be computed from one record alone, and what does the memory they need cost you operationally?
Every consumer whose question spans more than one event: "how many failed logins for this account in the last ten minutes", "did this order ever get paid", "is this the same event we already processed", "how long was this session". None of those can be answered by looking at a single record, so all of them make the job stateful.
The grain shifts from the event to the state entry: one row of state per key, or per key-and-window. Knowing what one state entry represents is what tells you how many there will be, which is the number that decides whether the job is operable at all.
Keep a dictionary in the operator. counts[account_id] += 1, seen.add(event_id), pending[order_id] = order. It is obvious, it is fast, and on a laptop against a day of test data it works perfectly.
The job restarts — a deploy, a node failure, a rebalance — and the dictionary is empty. Counts restart from zero, the seen-set forgets everything, and duplicates that were being suppressed flow through in a burst (Checkpointing).
- The job restarts — a deploy, a node failure, a rebalance — and the dictionary is empty. Counts restart from zero, the seen-set forgets everything, and duplicates that were being suppressed flow through in a burst (Checkpointing).
- The key space turns out to be unbounded.
seenholds every event id ever observed, grows for as long as the job runs, and the job dies from memory pressure weeks after the code that caused it was written (Streaming State). - Parallelism increases to handle a spike, partitions are reassigned, and an instance that now owns
account_42has never seen its history. The count for that account is wrong and there is no error anywhere (Consumer Groups and the Parallelism Ceiling). - Two events for the same order arrive on different partitions because the producer keyed by something else. The two halves of the join land on different instances and never meet — the join silently never matches (Event Keys and Partition Assignment).
- A replay from an earlier offset reprocesses records that already updated the state, so counts double. The job did exactly what it was asked; the state was not reset and the sink was not idempotent (Idempotent Data Pipelines).
What is actually happening
- A stateful operator is partitioned by key, and the engine guarantees that all records for one key reach the same instance. That is what makes per-key state safe without any locking — the concurrency has been removed by routing rather than by synchronisation (Shared Mutable State).
- State is held in a local store — process memory, or an embedded key-value store on local disk — because it is read and written on every record and a network round trip per record would set the throughput ceiling.
- Local state is not durable, so the engine periodically takes a consistent snapshot of the state together with the input positions that produced it, and writes it to durable storage. On restart it restores that pair. State and offsets moving together is what makes a restart resume rather than double-count (Checkpointing).
- The four canonical stateful operations are count/aggregate (state per key), window (state per key and window), join (a buffer of one side awaiting the other) and deduplicate (a set of seen identifiers with an expiry). Every other stateful operator is a variation on one of these.
- State must be bounded by something, and there are only three candidates: a time-to-live, a window that eventually closes, or a key space that is finite by construction. An operator with none of the three is a memory leak with a business justification (Memory Leaks: Growth That Does Not Come Back).
- Because state is keyed, rescaling means redistributing key ranges between instances, which means moving state. That is why a stateful job cannot be rescaled the way a stateless one can, and why parallelism changes are planned rather than automatic.
The four operations that need memory
It is worth being precise about which operations force state, because the list is short and each entry has a different bound. Once you can name which of the four an operator is, you can predict its state size, its failure mode and its recovery behaviour without knowing anything about the engine.
The pattern across all four is the same: state is keyed, its size is the number of live keys times the value size, and it is bounded by time or by nothing. When the answer to "what bounds this" is "nothing", you have found the operator that will end the job.
Deduplication is the one that catches people, because it looks like a filter. It is not: it requires remembering every identifier within its horizon, its key space is the identifier space rather than the entity space, and it is therefore usually the largest state in a pipeline that has one.
| Operation | What one state entry is | What bounds it | How it fails |
|---|---|---|---|
| Aggregate / count | One running value per key — a count, a sum, a min, a set of distinct values. | The number of live keys. Unbounded unless keys expire or the key space is finite. | Grows with entity count forever; a distinct-count state grows with cardinality inside each key as well. |
| Window | One partial aggregate per key and per open window. | Key count times the number of windows held open — which is why lateness allowance multiplies state directly. | A large lateness allowance or an overlapping window definition multiplies state by the overlap factor (Sliding Windows). |
| Join | A buffered record from one side, waiting for its counterpart. | Only the join's time bound. Without one, every unmatched record is retained forever. | An unbounded stream-stream join is a memory leak with a business justification (Stream Joins). |
| Deduplicate | One identifier that has been seen, plus the time it was seen. | The TTL. The key space is identifiers, so it is much larger than the entity space. | A TTL shorter than the real duplicate window lets duplicates through; a longer one costs memory proportional to the extension (Deduplication). |
Routing removes the concurrency problem, and creates a different one
The reason a stateful streaming operator can do a read-modify-write on a counter with no lock is not clever concurrency control. It is that the engine has routed every record for a given key to a single instance, so there is only ever one writer. The concurrency was eliminated by partitioning rather than by synchronisation (Shared Mutable State).
That trade is excellent and it has a bill. Keys must be assigned before the operator, which means either the producer keyed correctly or the engine had to shuffle. It means that all records for one key must be processed in sequence, so a hot key cannot be parallelised. And it means that changing parallelism changes which instance owns which key, which is why rescaling is a data movement problem.
It also means the two halves of a join must be keyed identically at the source. Two topics keyed differently do not converge just because the join condition says they should — the records land on different instances and the join simply never matches, silently, forever.
What one state entry is, and how many there will be
The question that predicts whether a stateful job is operable is not "how many events per second". It is "how many distinct state entries will exist, and what expires them". Answer that at design time and the job's operational profile is decided; skip it and you find out during an incident, when the checkpoint is taking longer than the checkpoint interval.
The table below walks the same pipeline and asks that question at every stateful stage. Note how the grain of state differs from the grain of the events flowing through: a stream of a thousand events per second about fifty accounts has fifty state entries, and a stream of fifty events per second with unique identifiers has a state entry per event until something expires it.
The last row is the one to internalise. Deduplication state is keyed by event identifier, so its size tracks *total events within the TTL* rather than entities — and it is therefore usually an order of magnitude more entries than everything else in the job combined, without appearing anywhere in the architecture diagram.
| Stage | One row is | Breaks if |
|---|---|---|
| Running count per account | One account, with its current count. | Accounts are never removed and the key is actually a session or a device, in which case the "account" count is really a visitor count and grows forever. |
| Five-minute tumbling sum per account | One account-window pair, with a partial sum. | Windows are held open by a lateness allowance, so the number of live windows per key is the allowance divided by the window size rather than one (Tumbling Windows). |
| Ten-minute sliding sum, sliding by one | One account-window pair, with the same account appearing in ten windows at once. | Nobody accounts for the overlap factor, and state is ten times the tumbling equivalent for the same data (Sliding Windows). |
| Session window per user | One open session for one user, holding everything since the session began. | A user never goes idle — a bot, a stuck client, a device that reconnects continuously — so the session never closes and one entry grows without limit (Session Windows). |
| Order-to-payment join buffer | One order awaiting a payment, or one payment awaiting an order. | The join has no time bound, so every order that is never paid is retained for the life of the job (Stream Joins). |
| Deduplication set | One event_id seen within the TTL. | The TTL is long, or absent. Entries scale with total event volume rather than with entity count, which makes this the largest state in most pipelines that have it (Deduplication). |
Multiply each row's entry count by its value size to get the state footprint, and that number — not the event rate — is what decides checkpoint duration, restore time and rescaling cost.
How to build it
Most important first.
- Declare the state explicitly rather than letting it appear as a local variable. Engines that make you name a state descriptor make you also answer "what is the key, what is the value, when does it expire" — which are the three questions that decide whether the job survives.
- Give every piece of state a bound before it is written: a TTL, a window, or a proof that the key space is finite. "We will add expiry later" is how unbounded state ships (Streaming State).
- Key by the thing the operation is actually about. A count per account must be keyed by account, and a join between orders and payments requires both streams keyed by the same order identifier — including at the producer, because repartitioning downstream costs a shuffle and breaks ordering for keys that move (Topics and Partitions).
- Make the sink idempotent regardless of what the engine promises about state. State recovery protects the *computation*; only the sink protects the *output*, and they are separate problems (Exactly-Once: Input Consumption, State Update, Output Write).
- Prefer state that can be rebuilt from the log over state that cannot. If a full replay reconstructs the state, a corrupt checkpoint is an inconvenience; if it cannot, a corrupt checkpoint is data loss (Replay from the Log).
- Emit state size and entry count as first-class metrics. Growth in state is the earliest observable form of most streaming incidents and the last thing anyone instruments.
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- Per-key single-threaded access: for a given key, one instance processes records one at a time, so read-modify-write on that key's state is atomic without locks. This holds only within one operator and only for that key.
- State durability equals checkpoint frequency. Anything computed since the last snapshot is recomputed after a restart, which is correct for the state and produces duplicate *effects* at any sink that was already written (At-Least-Once Delivery).
- Ordering within a key is preserved as long as the key stays on the same partition. It is not preserved across a repartition, and it never existed across keys (CDC Ordering and Transaction Boundaries).
- What is explicitly not promised: that state survives a change to the job's topology or key definition; that a restored checkpoint is compatible with new code; that rescaling is instantaneous; or that two keys' aggregates are consistent with each other at any instant.
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Reconcile the streaming aggregate against a batch recomputation over the same closed period from the raw events. This catches state lost across a restart, keys that were routed to the wrong instance, and windows that never closed (Reconciliation).
- It misses the case where the batch job and the stream share a definition bug, and it misses everything about the current open period — which is exactly where a state problem is newest.
- Add a bound check on state itself: assert that the number of state entries per operator stays within an expected range for the key cardinality you designed for. Unbounded growth is a data-shape problem long before it is a memory problem (Cardinality: The Label That Took Down Monitoring).
- Aggregations that emit on every record are as fresh as the record, and every emitted value is provisional — a downstream consumer sees a running number that changes constantly and has no way to know when it is final.
- Aggregations that emit on window close trade that for a defined answer: the consumer waits for the window plus the lateness allowance and then receives something stable. Which is right depends entirely on whether the consumer can tolerate a number that moves.
- State recovery adds a freshness cliff nobody plans for: after a failure, the job restores a snapshot and must reprocess everything since. During that catch-up the output is stale by a growing amount and then jumps — a shape that looks like a data incident to anyone watching a dashboard.
- Adding a field to a state value is often supported by the engine's state serialiser; changing the key, the window definition, the aggregation type or the operator graph is generally not. In practice, most meaningful changes mean starting with fresh state (Schema Evolution).
- The standard migration is dual-run: start a new job from an earlier offset with its own state and its own sink, let it catch up, compare its output with the running one, then move consumers. This makes log retention a hard constraint on how far back the new job can rebuild.
- A change to *what a key means* — for example, keying sessions by device rather than by user — invalidates all existing state silently. Nothing will fail; the numbers will simply be about a different thing from the deploy onwards (Semantic Changes).
- The normal path is restore-from-checkpoint: state plus offsets, restarted together. It is fast relative to a replay and only as good as the last snapshot.
- The fallback is a full replay from the log with empty state, which is slower but reconstructs state from first principles. It works only if the whole computation is deterministic given the log and the log still holds the range (Retention and Replay).
- The dangerous middle path is restoring a checkpoint and resetting offsets independently. Any combination other than the pair that was snapshotted together produces silently doubled or silently missing aggregates, and nothing reports it.
What can go wrong
- State grows without bound because a key space that was assumed finite is not — event ids, session ids, anonymous visitor ids are the classic three (Streaming State).
- A checkpoint that takes longer than the interval between checkpoints, so the job spends most of its time snapshotting and throughput collapses without any single component failing.
- A skewed key: one account with most of the traffic makes one instance hold most of the state and do most of the work, and adding parallelism does not help because the work is not divisible (Data Skew, Salting a Skewed Key).
- Restoring a checkpoint written by an incompatible version of the job, which either refuses to start (good) or starts and misinterprets the bytes (much worse).
- The mitigation failing: a state TTL short enough to bound memory quietly expires entries that a legitimately slow counterpart still needed, turning a memory fix into a correctness bug at the join (Stream Joins).
- "The engine gives exactly-once, so state and output are both safe." Engine-level exactly-once covers state updates with respect to input consumption — the same record does not update state twice. It says nothing about an external sink you wrote to, unless that write is transactional or idempotent (Exactly-Once: Input Consumption, State Update, Output Write).
- "State is just a cache; if we lose it we recompute." Only if a full replay can reconstruct it and the log still holds the range. State that depends on data older than retention is not a cache, it is the only copy.
- "We can add parallelism when we need it." Adding parallelism to a stateful job redistributes state, which takes time proportional to state size, which is largest exactly when you most need to scale.
- "Keying by user id gives even distribution." It gives distribution shaped like your users, which in every real product is heavily skewed. Even distribution is an assumption to be tested, not a property of hashing (Partition Cardinality).
Operating it
- State size in bytes and entry count, per operator and per instance. The per-instance split is what reveals skew; the aggregate hides it.
- Checkpoint duration, checkpoint size and failed-checkpoint count over time. A rising duration is the single best leading indicator of a streaming job that is about to become unoperable.
- Restore duration after each restart, because it is the number that decides your recovery time objective and it is almost never measured until an incident (RPO & RTO).
- Per-key hot-spot detection: the top keys by record count and by state size, which turns a vague "the job is slow" into a specific key (Hot Keys: When Aggregate Metrics Hide a Saturated Node).
- At 10x throughput with the same key space, state barely changes and the job scales with parallelism like a stateless one. This is the comfortable case and it is why the problem is usually a surprise.
- At 10x key cardinality, state grows tenfold whatever the throughput does. Checkpoints get slower, restores get slower, and the job crosses from "restart in a moment" to "restart is an incident".
- At 100x, the state store's own characteristics start to dominate — how it handles compaction, how random its access pattern is, whether it fits in memory — and the job takes on the operational profile of a database, because that is what it now is (LSM Trees: Why Some Engines Favour Writes).
- State is memory, local disk, checkpoint storage and checkpoint network traffic. It is driven by distinct keys times retained time range times value size, and notably not by throughput — a low-volume job with a huge key space costs more here than a high-volume one with a small one.
- Checkpointing costs a periodic write proportional to state size, so state growth costs twice: once to hold it and once, repeatedly, to snapshot it. Incremental checkpointing changes the shape of the second term but not the first.
- Rescaling costs a state redistribution, which is a bulk transfer proportional to state size. This is the reason a stateful job is expensive to scale reactively and should be sized for its peak.
- Stateful operators buy answers that span events and cost you free rescaling, free redeploys and simple recovery. That is the trade, and it should be made deliberately per operator rather than inherited by an entire job.
- A TTL bounds state and introduces a correctness boundary: anything that needed to be remembered longer is now silently wrong. Choosing the TTL is choosing which correctness you are giving up.
- Frequent checkpoints reduce the work lost on failure and increase steady-state cost and latency. Infrequent ones do the reverse. There is no setting that is good at both, and the right one depends on how expensive reprocessing is for your job.
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALKeyed partitioning, local state, snapshot-with-offsets and the four canonical stateful operators are shared by every stream processor; the argument for each of them follows from the physics of reading state per record rather than from any product decision.
- ENGINE-SPECIFICFlink snapshots via barriers flowing through the dataflow and supports incremental checkpoints with a disk-backed state store; Spark Structured Streaming checkpoints per micro-batch to a checkpoint location; Kafka Streams keeps local stores backed by compacted changelog topics, so its recovery mechanism is a topic replay rather than a snapshot restore. Recovery time and rescaling behaviour differ accordingly.
- SCALE-SPECIFICBelow a key cardinality that fits comfortably in memory, most of this lesson is theory and a plain dictionary really is fine. Above it, state store behaviour, checkpoint duration and restore time become the job's dominant operational properties.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns what a consistent snapshot across many operators actually requires, and why taking one without stopping the world is a non-trivial algorithm rather than a configuration flag.