KeyBy and key groups
Why is keyBy a live shuffle through a fixed indirection table, and why is its size permanent?
Repartitioning and keyBy
A keyBy sends each record to the instance that owns its key. This is the streaming version of a shuffle (redistributing records between machines). It differs from a batch shuffle in three ways that matter:
- There is no barrier. A batch shuffle can wait until a stage ends, then write and sort the data. A stream cannot wait: records must be forwarded as they arrive.
- The connections are live and continuous. The all-to-all connection set (every upstream instance can send to every downstream instance) is kept open, rather than being created and torn down for each stage.
- It is the channel along which backpressure propagates (Backpressure propagation).
Key groups: the indirection
The mapping is not hash(key) mod parallelism. It is
and contiguous ranges of key groups are assigned to instances.
Why the indirection exists. A direct mod N mapping remaps nearly every key when N changes, which is why consistent hashing exists. Key groups are the cheaper answer here, logical shards over physical instances, because the assignment is centrally coordinated and need not be stable under arbitrary membership changes.
The permanent decision
max_parallelism is fixed at job creation and caps scaling forever. Too few groups cap you: with 128 groups you can never run more than 128 instances. Too many groups add per-group metadata and file overhead at checkpoint time. You choose it before you know the load.
Per-record cost and the buffer timeout
Per-record cost across the shuffle is serialization plus network transfer. The buffer timeout setting is the explicit latency-versus-throughput knob (the fixed-cost law): records are collected in a network buffer, which is sent when it is full (best throughput) or when the timeout expires (bounded latency), whichever comes first.