Definition
- Keyed state – Partitioned by the stream's key, accessible only within the context of the current key; each key has its own independent slice, exposed through
ValueState,ListState,MapState,ReducingState, andAggregatingState. - Operator state – Scoped to a single parallel operator instance rather than to a key, exposed through
ListState,UnionListState, andBroadcastState; typically used by sources and sinks. - ValueState – Single value per key, e.g. a running account balance.
- ListState – List of values per key, e.g. buffered events.
- MapState – Key-value map per key, e.g. sub-keyed lookups.
- BroadcastState – Map-based operator state replicated identically to every parallel instance for rules or config.
- Checkpointing – Both keyed and operator state are snapshotted into checkpoints for exactly-once fault tolerance.
Keyed state is like giving each customer at the National Bank of India their own passbook: when a transaction for Rajesh arrives, the clerk opens only Rajesh's passbook and never touches Priya's, because the state is strictly partitioned per customer. Operator state is like the clerk's own desk notebook that belongs to the worker, not to any customer, for example a note of "the last page number I read from the incoming ledger file."
Understanding the Concepts
What is keyed state?
Keyed state is only available on a keyed stream, after a keyBy. Flink automatically scopes the state to the current key, so your code reads and writes "the state" and Flink routes it to the right key's slice.
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class BalanceTracker
extends KeyedProcessFunction<String, Transaction, String> {
private transient ValueState<BigDecimal> balanceState;
@Override
public void open(Configuration params) {
ValueStateDescriptor<BigDecimal> desc =
new ValueStateDescriptor<>("balance", BigDecimal.class);
balanceState = getRuntimeContext().getState(desc);
}
@Override
public void processElement(Transaction txn, Context ctx,
Collector<String> out) throws Exception {
BigDecimal current = balanceState.value();
if (current == null) current = BigDecimal.ZERO;
BigDecimal updated = current.add(txn.getAmount());
balanceState.update(updated);
out.collect(txn.getAccountId() + ": " + updated);
}
}
Input: ACC-001 +50000, ACC-002 +30000, ACC-001 +20000
Output: ACC-001: 50000
ACC-002: 30000
ACC-001: 70000 (running balance updated)
What are the keyed state primitives?
Flink provides five keyed state types, each suited to different patterns.
| State Type | Stores | Use Case |
|---|---|---|
| ValueState | Single value per key | Running balance, last seen timestamp |
| ListState | List per key | Buffered events, transaction history |
| MapState | Map per key | Per-account sub-keyed lookups |
| ReducingState | Single value, incrementally combined | Running sum |
| AggregatingState | Like Reducing but input/output differ | Custom aggregation |
// ListState: buffer recent transactions per account
ListStateDescriptor<Transaction> listDesc =
new ListStateDescriptor<>("recentTxns", Transaction.class);
ListState<Transaction> recentTxns =
getRuntimeContext().getListState(listDesc);
recentTxns.add(txn); // append to list for current key
// MapState: per-account merchant spend tracking
MapStateDescriptor<String, BigDecimal> mapDesc =
new MapStateDescriptor<>("merchantSpend",
String.class, BigDecimal.class);
MapState<String, BigDecimal> merchantSpend =
getRuntimeContext().getMapState(mapDesc);
merchantSpend.put(txn.getMerchantId(), newTotal);
What is operator state?
Operator state is scoped to one parallel subtask and does not depend on a key. It is used by sources and sinks for per-worker plumbing like offset tracking and buffer management.
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
public class OffsetTrackingSource
implements CheckpointedFunction {
private transient ListState<Long> offsetState;
private long currentOffset = 0;
@Override
public void initializeState(FunctionInitializationContext ctx)
throws Exception {
ListStateDescriptor<Long> desc =
new ListStateDescriptor<>("offsets", Long.class);
offsetState = ctx.getOperatorStateStore()
.getListState(desc);
// Restore offset from checkpoint
for (Long offset : offsetState.get()) {
currentOffset = offset;
}
}
@Override
public void snapshotState(FunctionSnapshotContext ctx)
throws Exception {
offsetState.clear();
offsetState.add(currentOffset);
}
}
How is keyed state different from operator state?
| Aspect | Keyed State | Operator State |
|---|---|---|
| Scope | Per key (after keyBy) | Per parallel subtask |
| Access | Within key context only | Within operator instance |
| Rescaling | Redistributed by key group | ListState: round-robin; Union: full copy |
| Primitives | Value, List, Map, Reducing, Aggregating | ListState, UnionListState, BroadcastState |
| Use case | Per-entity: balances, counters, sessions | Per-worker: offsets, buffers, rules |
// Keyed state: per-account balance (after keyBy)
stream.keyBy(txn -> txn.getAccountId())
.process(new BalanceTracker()); // ValueState per account
// Operator state: source offset tracking (no keyBy)
// Source reads partitions and remembers its own read position
What is BroadcastState?
BroadcastState is a special map-based operator state used with a broadcast stream. A small, slowly-changing dataset like fraud rules is replicated identically to every parallel instance.
import org.apache.flink.api.common.state.MapStateDescriptor;
MapStateDescriptor<String, FraudRule> ruleDescriptor =
new MapStateDescriptor<>("rules", String.class, FraudRule.class);
BroadcastStream<FraudRule> broadcastRules =
ruleStream.broadcast(ruleDescriptor);
// Every parallel instance gets an identical copy of the rules
DataStream<String> alerts = txnStream
.keyBy(txn -> txn.getAccountId())
.connect(broadcastRules)
.process(new FraudScreeningFunction());
How does rescaling work?
Keyed state rescales gracefully by redistributing key groups across the new parallelism. Operator state requires explicit handling: ListState redistributes round-robin, UnionListState gives every instance the full set.
// Keyed state: parallelism 4 → 8
// Key groups automatically redistributed
// ACC-001 moves from task 2 to task 5 with its state intact
// Operator ListState: parallelism 2 → 4
// Elements split round-robin: [A,B,C,D] → task0:[A,C] task1:[B,D]
// Operator UnionListState: parallelism 2 → 4
// Every task gets full set: [A,B,C,D] → all tasks get [A,B,C,D]
When do you use each?
Use keyed state for anything naturally per-entity: per-account balances, per-card fraud counters, per-customer sessions. Use operator state for per-worker plumbing: source read positions, sink buffers, and broadcast rules.
// Keyed state: per-entity logic
stream.keyBy(txn -> txn.getAccountId())
.process(new BalanceTracker()); // ValueState per account
stream.keyBy(txn -> txn.getCardId())
.process(new FraudCounter()); // MapState per card
// Operator state: per-worker plumbing
// Source: remembers offsets (ListState)
// Sink: buffers rows for batch commit (ListState)
// Rules: broadcast to all instances (BroadcastState)
What are common mistakes?
The most common mistake is using operator state for per-entity logic, which does not partition by key and breaks when parallelism changes. Another mistake is forgetting that BroadcastState must be small because it is fully replicated.
// WRONG -- using operator state for per-account balance
// Not partitioned by key! All accounts on same subtask.
// Rescaling loses data.
// CORRECT -- use keyed state after keyBy
stream.keyBy(txn -> txn.getAccountId())
.process(new BalanceTracker()); // ValueState per account
What should a senior engineer know?
Keyed state is far more common and rescales gracefully by key group. Operator state requires you to think about how contents redistribute on parallelism changes. BroadcastState must be kept small because it is fully replicated. Both kinds are snapshotted into checkpoints for exactly-once recovery, and the choice of state backend determines whether state lives on the JVM heap or in an embedded on-disk store.
Visual Overview
Flink State Categories
Keyed State (after keyBy):
┌─────────────────────────────────────────┐
│ Key: ACC-001 │
│ ┌──────────────────────┐ │
│ │ ValueState: 150000 │ (balance) │
│ │ ListState: [T1,T2] │ (history) │
│ │ MapState: {M1:5000} │ (merchants) │
│ └──────────────────────┘ │
│ Key: ACC-002 │
│ ┌──────────────────────┐ │
│ │ ValueState: 80000 │ │
│ │ ListState: [T3] │ │
│ └──────────────────────┘ │
└─────────────────────────────────────────┘
Each key has isolated state. Rescales by key group.
Operator State (per subtask):
┌──────────────┐ ┌──────────────┐
│ Subtask 0 │ │ Subtask 1 │
│ ListState: │ │ ListState: │
│ [offset=42] │ │ [offset=87] │
│ BcastState: │ │ BcastState: │
│ {rule1,rule2}│ │ {rule1,rule2}│
└──────────────┘ └──────────────┘
Per-worker. BroadcastState identical on all.
Key Takeaway
Flink state is the memory an operator keeps across events so it can compute results that depend on history, and Flink makes that state fault-tolerant by snapshotting it into checkpoints. There are two categories. Keyed state is partitioned by the stream's key and is only accessible after a keyBy, within the context of the current key, so every key gets its own independent slice. Flink provides several primitives: ValueState for a single value per key, ListState for a list, MapState for a map, ReducingState for incremental combination, and AggregatingState for flexible aggregation.
Operator state is scoped to a single parallel instance rather than to a key. ListState holds a per-subtask list redistributed round-robin on rescaling, UnionListState gives every instance the full set on restore, and BroadcastState replicates a small dataset identically to every instance. You implement operator state through the CheckpointedFunction interface.
The rule for when to use each is straightforward. Keyed state is the default for per-entity logic: per-account balances, per-card fraud counters, per-customer sessions. Operator state is for per-worker plumbing: source read positions, sink buffers, and broadcast rules.
At the National Bank of India the running-balance job used keyed ValueState for each account's balance so that a transaction only ever touched that account's slice and the job rescaled cleanly. It used operator ListState in the source to remember partition offsets, and BroadcastState for fraud rules replicated to every instance.
Remember one sentence: Keyed state is per-entity memory partitioned by key for business logic, operator state is per-worker plumbing for sources, sinks, and broadcasts, and both are checkpointed for exactly-once recovery.
---