Keyed State and Operator State

Analogy categoryAnalogies

What Is State in Flink?

Flink applications that go beyond stateless transformations rely on managed state to remember information across events — running counts, session windows, deduplication sets, or machine state for a CEP pattern. Flink exposes two distinct state primitives, keyed state and operator state, each with different scoping rules, APIs, and behavior when a job is rescaled. Choosing the right one determines whether your application can correctly resume after a failure or a parallelism change.

🏏

Cricket analogy: Just as a scorer tracks separate running totals for each batsman at the crease rather than one combined number for the whole team, Flink needs a way to keep per-entity state distinct from state that belongs to the whole operator instance.

Keyed State: One State Instance Per Key

Keyed state is scoped to the key extracted by a keyBy() operation: Flink guarantees that every record with the same key is routed to the same parallel subtask, so a ValueState, ListState, or MapState declared inside a RichFunction or KeyedProcessFunction transparently holds one independent value per key. Internally, keys are hashed into a fixed number of key groups (maxParallelism determines this count), and each parallel subtask owns a contiguous range of key groups, which is what makes state repartitioning during rescaling deterministic rather than requiring a full shuffle.

🏏

Cricket analogy: MS Dhoni's per-match strike rate is tracked as an isolated running number keyed to 'Dhoni', completely separate from Virat Kohli's strike rate, even though both numbers live in the same scorebook system — that's keyed state's per-key isolation.

java

public class PurchaseCounter extends KeyedProcessFunction<String, Purchase, Tuple2<String, Long>> {
    private transient ValueState<Long> countState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor =
            new ValueStateDescriptor<>("purchaseCount", Long.class, 0L);
        countState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Purchase purchase, Context ctx,
                                Collector<Tuple2<String, Long>> out) throws Exception {
        Long current = countState.value();
        long updated = current + 1;
        countState.update(updated);
        out.collect(Tuple2.of(purchase.getCustomerId(), updated));
    }
}

Operator State: Per-Parallel-Task State

Operator state, in contrast, is not tied to any key — it belongs to a single parallel instance of an operator, regardless of how many distinct keys or records pass through it. It's most commonly used by source and sink connectors: the Kafka consumer, for instance, uses a ListState to track the (partition, offset) pairs it owns, since offset tracking has nothing to do with the content of individual records. Operator state is accessed through the CheckpointedFunction interface rather than getRuntimeContext().getState(), and it comes in two flavors — union state, where every restored task receives the full union of all state from all tasks at restore time, and even-split (list) state, where each parallel task receives an even chunk.

🏏

Cricket analogy: A ground curator's pitch-moisture reading isn't tied to any individual batsman — it belongs to the pitch itself for that match, just as operator state belongs to the parallel task instance, not to any key.

Implement CheckpointedFunction and override snapshotState() and initializeState() to use operator state directly; most application code never needs this because Flink's built-in connectors already manage operator state for source offsets and sink transaction buffers.

Redistribution During Rescaling

When a job is restarted with a different parallelism, keyed state redistributes deterministically because key groups are reassigned to the new set of subtasks based on hash ranges — no data is lost or duplicated, and each key's state lands on exactly one new subtask. Operator state redistributes differently depending on which redistribution strategy the connector chose at snapshot time: even-split state divides the accumulated list roughly evenly across the new parallelism, while union state gives every new subtask the complete combined list, which is why Kafka source implementations historically used union state for partition assignment to let each subtask independently decide, via consistent hashing, which partitions it now owns.

🏏

Cricket analogy: When a tournament expands from eight teams to ten, the fixture scheduler deterministically reassigns each existing team to a new group slot without losing any team's match history, just as key groups are reassigned to new subtasks on rescale.

Broadcast state, used with a BroadcastProcessFunction, is a distinct third category — identical, read-only copies of a MapState replicated to every parallel task, most often used to fan out low-throughput control-stream data like feature flags or fraud rules to all instances of a keyed stream.

Keyed state is scoped per key after keyBy() and is accessed via ValueState, ListState, MapState, ReducingState, or AggregatingState inside a RichFunction.
Keys are hashed into key groups, and each parallel subtask owns a contiguous range of key groups, enabling deterministic redistribution on rescale.
Operator state belongs to a parallel task instance, not to a key, and is implemented via the CheckpointedFunction interface.
Operator state connectors choose between even-split (ListState) and union (UnionListState) redistribution strategies at snapshot time.
Kafka source and sink connectors are the canonical example of operator state usage — tracking partition offsets and transactional buffers.
Broadcast state is a separate, third kind of state: read-only, identical copies replicated to every parallel task.
Choosing keyed vs. operator state incorrectly can break rescaling correctness, so connector authors must pick the right redistribution semantics deliberately.