Keyed State Layout and the Cost of Keeping It

Every aggregation, join, and deduplication in a stream pipeline is a stateful operator, and the state is the part of the job that fails. Compute scales linearly with records and can be added; state grows with the number of distinct keys and with the time range held, and cannot be added without moving it. The engineering question is therefore not how to avoid state but where to put it, and the answer determines both the memory footprint and the recovery time.

The unit is the key, not the record

A keyed operator partitions its state by a key derived from the record, and each key has its own entry. For a count, the entry is an integer. For a sum, an accumulator. For a windowed aggregation, one entry per key per open window, so the state size is the number of distinct keys times the number of open windows times the per-entry size.

That multiplication is the whole cost model. A job with ten million distinct keys and a one-hour window with a one-minute hop has six hundred million window entries at any moment, because each event lands in sixty windows and each window stays open for an hour. Even at a few dozen bytes per entry that is far beyond a heap. Nothing about the record rate indicates this; the pipeline can be handling a modest number of events per second and still carry an enormous key space, because the keys are what accumulate.

The classic instance is unbounded key cardinality: a key derived from a user identifier, a session identifier, or a request identifier. A job keyed by session sees a new key for every session and never revisits it, so state grows monotonically with traffic and no window ever closes to release it. The operator is not leaking in the sense of a bug; it is doing exactly what it was asked, over a key space that has no bound. The failure arrives as a slow memory climb, then an out-of-memory kill, then a restart that replays the same climb.

The fix is a state TTL or an explicit eviction. A TTL declares that a key’s state is discarded after a period of inactivity, which is right for session state and wrong for anything that must be exact over all history. Eviction is a policy decision about what correctness the pipeline gives up, and it belongs in the design rather than in a tuning parameter discovered after the first crash.

Backends: heap, disk, and the hybrid

An in-memory backend keeps each key’s state as an object on the JVM or runtime heap. Access is a hash lookup, so reads are fast and constant, and the cost is that the state must fit in the heap alongside everything else the operator does. A heap backend also makes checkpoints expensive, because the snapshot has to serialize the whole object graph, and the serialization is a stop-the-world-ish operation proportional to state size.

A disk-backed backend, typically an embedded LSM store, keeps state in local files with a block cache in front. Reads that hit the cache cost the same as a hash lookup; reads that miss go to disk. The advantage is that state can exceed heap by orders of magnitude, and checkpoints become incremental: the store’s files are copied rather than the live objects serialized, so a checkpoint captures only the files created since the previous one. The disadvantage is the read path — an LSM lookup may consult several levels of sorted files before finding the key, and the block cache is a fixed memory budget that the working set may exceed.

The choice follows from the ratio of state size to available memory. If the state fits comfortably, the heap backend is simpler and faster. If it does not, the disk backend is the only option that works, and the tuning that matters is the cache size relative to the hot key set. A job whose keys are uniformly accessed has no hot set and every lookup goes to disk; a job whose keys follow a skewed distribution — a few very active keys and a long tail — gets most lookups from the cache and pays disk only for the tail.

// Per-key state with an inactivity TTL. The descriptor is registered
// once at operator construction; the value is per key at runtime.
StateTtlConfig ttl = StateTtlConfig
        .newBuilder(Time.hours(6))
        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
        .build();

ValueStateDescriptor<Long> desc =
        new ValueStateDescriptor<>("session-count", Long.class);
desc.enableTimeToLive(ttl);

The update type is the setting that matters for cost. OnCreateAndWrite extends the TTL only when the value is written, so a key that is read frequently but written rarely still expires on schedule; extending it on every access keeps hot keys alive indefinitely, which is the behaviour wanted for a genuine session and wrong for a cache of derived values.

Checkpoint size follows state size

A checkpoint must capture every piece of state that the operator would need to resume correctly, which includes the aggregates, the watermark position, and the timers. Timers deserve attention because they are state too: a window with a timer set to fire at its end has one timer entry per key per window, and a job with hundreds of millions of open windows has hundreds of millions of timers. In a heap backend those timers are objects with their own overhead, and their count can exceed the aggregate entries they guard.

The checkpoint interval interacts with this. A checkpoint every ten seconds over a large state means the serializer runs against a moving target continuously, and the job spends a growing fraction of its time snapshotting. A checkpoint every ten minutes means recovery replays ten minutes of data, and the replay is bounded by the source’s retention. The interval is chosen against the acceptable recovery time, and the acceptable recovery time is usually set by the downstream’s tolerance for a gap, not by the pipeline’s own preference.

Checkpoint Barriers and Aligned Snapshots covers how the snapshot is taken without stopping the stream. The relevant point here is that the snapshot’s size is the state’s size, so every decision that grows state also grows recovery time. A job that quadruples its retained window count quadruples its checkpoint and its restart duration, and the restart may exceed the source’s retention, which turns a recoverable failure into an unrecoverable one.

Keyed state and parallelism

State is partitioned by key, and the partition is a function of the key and the operator’s parallelism. Changing the parallelism changes the assignment, so state must be redistributed — which is what Rescaling Keyed State is about. The design consequence is that the key’s hash must be stable across runs, which rules out hash functions seeded per process and rules out keys whose serialized form changes between versions.

A second consequence is that a hot key cannot be split. A key with ten times the traffic of the next one lands entirely on one subtask, and that subtask becomes the bottleneck while the others idle. The remedy is to salt the key — append a synthetic suffix, aggregate per salt, then combine — which multiplies the state and requires a second aggregation stage. It is the standard answer and it is always a change to the job’s structure, not a configuration flag.

Reading the numbers

The quantities that predict trouble are the number of distinct keys, the number of open windows per key, and the retained duration. Their product, times the per-entry size, is the state footprint. Comparing that against the memory budget gives the ratio that decides the backend, and comparing it against the checkpoint interval gives the serialization rate. Both comparisons are arithmetic on values the job can report, which is why a stateful pipeline should expose its key count and its open-window count as metrics rather than only its throughput.

The failure mode to watch for is the one with no error message: state grows because the key space grows, the backend spills more to disk, lookups slow, throughput falls, backpressure builds, and the source’s lag increases until the watermark stalls. Every stage of that sequence is a symptom of the first, and by the time it is visible the fix is a change to the key definition, which is a job rewrite rather than a tuning adjustment.