Rescaling Keyed State Without Replaying History

Changing a job’s parallelism from eight subtasks to sixteen looks like a configuration edit. For a stateless operator it is one: the runtime starts more instances and routes records among them. For a keyed operator it requires redistributing the state, because each key’s aggregate lives in the subtask that was hashing it, and the hash depends on the parallelism. Doing this by replaying the source from the beginning is correct and can take days. The indirection that makes it a file move is the most consequential design choice in a stateful streaming runtime.

The problem in one sentence

A keyed operator assigns key k to subtask hash(k) mod p, where p is the parallelism. Change p and the modulus changes, so almost every key moves to a different subtask. If state is not moved with it, each subtask resumes with aggregates for keys it no longer owns and without aggregates for keys it now owns, and the first is a leak while the second is a wrong answer.

The naive remedies are both bad. Replaying from the source rebuilds the state correctly and costs the full history, which for a job with a month of retained state is a month of reprocessing. Copying every key’s state to every new subtask is correct only if each subtask can filter to its own keys, and the copy is p_new times the state size, which for a large state is prohibitive.

Key groups: a stable intermediate index

The construction that avoids both is a two-level mapping. Keys are first hashed into a fixed number of key groups — a count fixed when the job is first deployed, typically declared as its maximum parallelism — and key groups are then assigned to subtasks. The assignment is a range partition over the group index: with g groups and p subtasks, subtask i owns groups [i*g/p, (i+1)*g/p).

Because the group count is fixed, a key’s group is stable across rescaling, and a key’s state moves only when its group’s owning subtask changes. Changing p reassigns whole groups, so the state transfer is a set of contiguous group ranges moving between subtasks rather than a per-key shuffle. State is laid out by group range within each subtask’s store, so the transfer is a copy of the ranges that changed hands.

g = 128 groups, p = 8 subtasks  -> 16 groups per subtask
g = 128 groups, p = 16 subtasks ->  8 groups per subtask

rescaling 8 -> 16: each old subtask's 16 groups split into two halves,
one kept and one sent to a new subtask. Half the state moves.

The numbers show the property that makes this practical: the volume moved is bounded by the fraction of groups whose owner changed, and for a doubling of parallelism that fraction is one half. It does not depend on the key count, and it does not require reading the data.

The group count is the parameter that determines the granularity. Too few groups and the assignment is coarse: with g = 10 and p = 16, some subtasks own one group and some own none, so the effective parallelism is ten. Too many groups and the per-group state files become numerous and small, which costs filesystem metadata and per-file overhead more than it costs bytes. A group count in the low hundreds supports parallelism up to that count with reasonable granularity, and the choice is made once.

What the runtime does during a rescale

The sequence is: stop the job at a checkpoint, read the checkpoint’s state manifest, compute the new group-to-subtask assignment, transfer the groups that changed owner, start the new subtasks with their assigned groups, and resume from the checkpoint’s source positions.

The transfer can be done from the checkpoint rather than from the live subtasks, which is the reason rescaling is safe. The checkpoint is immutable once written, so the new topology reads it without coordinating with the old one, and a failure during the transfer leaves the checkpoint intact and the job restartable at the old parallelism.

Two properties make the resume correct. The first is that source positions are per-partition and unaffected by operator parallelism; the operator topology changes but the read positions do not. The second is that the watermark is part of the checkpoint and is restored as it was, so windows that were open remain open and closed ones stay closed. Checkpoint Barriers and Aligned Snapshots is what makes the checkpoint consistent enough to be the source of the transfer.

The cost is a pause. The job is stopped for the duration of the state copy, which for a large state on slow storage can be minutes. An alternative is to run the new topology alongside the old one and migrate, but the two would then produce duplicate output for the overlapping period, so the pause is the simple correct answer and is what most runtimes implement.

When state cannot be key-group partitioned

Three cases defeat the scheme, and each has its own remedy.

The first is unkeyed operator state, which is a single value per subtask rather than per key: a running count, a file offset, a connection handle. This state is not partitioned by key, so it cannot be split. The runtime’s options are to broadcast the whole value to every new subtask, which is correct only for state that is genuinely replicated, or to require the operator to declare a redistribution function. An unkeyed operator whose state is a scalar total cannot be rescaled without recomputation, because half of a total is not a meaningful half.

The second is state whose per-key entries reference each other. A session window’s merged state, or a join’s paired buffers, may hold entries that are only meaningful together. If the key group boundary falls between them, the halves end up in different subtasks and the relationship is lost. The runtime cannot detect this, so the operator’s keying must be chosen so that related entries share a key — a session keyed by user rather than by session identifier, a join keyed by the entity on both sides.

The third is a key space with extreme skew. Rescaling redistributes keys uniformly by group, but if one key carries most of the traffic, no assignment of groups moves that traffic to more than one subtask. The hot key remains a bottleneck at every parallelism, and the remedy is salting from Keyed State Layout and the Cost of Keeping It rather than rescaling.

Rescaling as a routine operation

The value of the key-group design is not that rescaling is possible but that it is cheap enough to be routine. A pipeline whose parallelism is fixed at deployment time is sized for a peak that may occur once a year, and runs over-provisioned for the rest. A pipeline that can rescale in minutes can be sized for the current load, which changes the cost model of operating it.

The precondition is that the job’s state is keyed and that the key groups were configured at creation. A job whose state is unkeyed, or whose keys are unstable across versions, cannot use the mechanism and must be restarted from scratch. The decision is made once, at the first deployment, and it constrains every later operation — which is why the group count and the key definition deserve more attention than they usually get when a job is first written.

The remaining limitation is that rescaling is not free even in the good case. The pause is a service interruption, the transfer is I/O proportional to the moved state, and the checkpoint that anchors the operation must be complete. A pipeline that rescales frequently is one whose parallelism is being used as a load-control knob, and at that point the more economical answer is usually to fix the skew or the key space instead of resizing around it.