What keyBy() Does
Calling keyBy() on a DataStream logically partitions it into disjoint KeyedStream partitions based on the result of a key selector function, and Flink guarantees that all records sharing the same key are always routed to the same parallel subtask. This routing is achieved by hashing the key and mapping it into one of a fixed number of key groups, which are then assigned to subtasks, making keyBy() the foundation for any per-entity aggregation such as counting events per user or computing a rolling sum per sensor.
Cricket analogy: keyBy() is like sorting every ball bowled in an IPL season by bowler name so that every delivery from Jasprit Bumrah always lands in the same analysis bucket, regardless of which match it came from.
Keyed State
Once a stream is keyed, operators can use keyed state primitives — ValueState, ListState, MapState, and ReducingState — which are automatically scoped to the current key and only accessible from within a keyed context such as a RichFlatMapFunction applied to a KeyedStream. Flink manages this state per key transparently, including checkpointing it and redistributing it if the job is rescaled, so application code simply calls state.value() or state.update() without ever manually tracking which key is active.
Cricket analogy: Keyed state is like a scorer keeping a running strike-rate tally per batsman on a physical scorecard — each player has their own dedicated cell that only updates when that specific batsman is on strike.
public class RunningTotal extends RichFlatMapFunction<Tuple2<String, Double>, Tuple2<String, Double>> {
private transient ValueState<Double> total;
@Override
public void open(OpenContext ctx) {
ValueStateDescriptor<Double> descriptor =
new ValueStateDescriptor<>("running-total", Double.class, 0.0);
total = getRuntimeContext().getState(descriptor);
}
@Override
public void flatMap(Tuple2<String, Double> event, Collector<Tuple2<String, Double>> out) throws Exception {
double updated = total.value() + event.f1;
total.update(updated);
out.collect(Tuple2.of(event.f0, updated));
}
}
// Applying it requires keyBy() first:
DataStream<Tuple2<String, Double>> sales = ...;
sales.keyBy(t -> t.f0)
.flatMap(new RunningTotal())
.print();Key Selection and Hash Partitioning
The key extracted by a KeySelector must be deterministic and rely on a consistent, well-behaved hashCode() and equals() implementation, because Flink hashes the key to assign it into one of maxParallelism key groups, and an inconsistent key would cause records for what should be the same logical entity to land on different subtasks. Using mutable objects, or objects whose hashCode changes over time, as keys is a common source of subtle bugs where state appears to silently reset or split.
Cricket analogy: A bad key selector is like a scorer using a player's current jersey number as their identity — if a player changes numbers mid-season, their earlier stats suddenly stop matching, splitting one career into two records.
The maxParallelism setting (default 128) determines the fixed number of key groups a job's keyspace is divided into, independent of the current operator parallelism. This decoupling is what allows Flink to rescale a job's parallelism later without redistributing every individual key — only whole key groups need to move.
Skew and Hot Keys
Because keyBy() forces a network shuffle so that records for each key converge onto one subtask, an uneven distribution of keys — commonly called data skew — can overload a single subtask with a disproportionate share of the traffic while other subtasks sit idle. This is especially common with 'hot keys', such as one viral product ID receiving far more events than any other, and it manifests as backpressure and an inability to scale out further simply by increasing parallelism, since one key can never be split across multiple subtasks.
Cricket analogy: A hot key is like every fan's question during a post-match press conference being directed at Virat Kohli alone — one player is overwhelmed with attention while the rest of the team sits idle, no matter how many reporters are in the room.
A single hot key can never be processed by more than one subtask, so increasing overall job parallelism does nothing to relieve a skewed key's bottleneck. Mitigations include salting the key with a random suffix to spread a hot key's load across multiple sub-keys and aggregating in two stages, or pre-aggregating with a local combiner before the keyed shuffle.
- keyBy() logically partitions a DataStream so all records with the same key go to the same subtask.
- Keyed state (ValueState, ListState, MapState, ReducingState) is automatically scoped per key.
- Key selectors must be deterministic and rely on stable hashCode/equals implementations.
- maxParallelism fixes the number of key groups independent of current operator parallelism, enabling rescaling.
- Data skew and hot keys can overload a single subtask regardless of total job parallelism.
- A single key can never be split across multiple subtasks, so hot-key mitigation requires key salting or pre-aggregation.
- Keyed streams are the prerequisite for correct per-entity aggregations and windowed computations.
Practice what you learned
1. What guarantee does keyBy() provide?
2. Why is using a mutable object with an unstable hashCode as a key problematic?
3. What does maxParallelism control in a keyed Flink job?
4. Why can't increasing job parallelism alone fix a hot key bottleneck?
5. Which state type would you use to track a single running numeric total per key?
Was this page helpful?
You May Also Like
DataStream Basics
An introduction to Flink's DataStream API, the core abstraction for processing unbounded and bounded streams of data in real time.
Transformations: Map, Filter, FlatMap
How to use Flink's core stateless transformation operators — map, filter, and flatMap — to reshape and clean streaming data.
Connectors: Sources and Sinks
An overview of Flink's connector ecosystem for reading from and writing to external systems like Kafka, files, and databases.
Related Reading
Related Study Notes in Programming
Browse all study notesApache Spark Study Notes
Programming · 30 topics
ProgrammingHadoop Study Notes
Programming · 30 topics
ProgrammingSnowflake Study Notes
Programming · 30 topics
ProgrammingApache Airflow Study Notes
Programming · 30 topics
Programmingdbt (Data Build Tool) Study Notes
Programming · 30 topics
ProgrammingPowerShell Study Notes
Programming · 30 topics