Stephan Ewen created FLINK-21695:
------------------------------------
Summary: Increase default value for number of KeyGroups
Key: FLINK-21695
URL: https://issues.apache.org/jira/browse/FLINK-21695
Project: Flink
Issue Type: Improvement
Components: Runtime / Checkpointing, Runtime / State Backends
Reporter: Stephan Ewen
Fix For: 1.13.0
The current calculation for the number of Key Groups (max parallelism) leads in
many cases to data skew and to confusion among users.
Specifically, the fact that for maxParallelisms above 128, the default value is
set to {{roundToPowerOfTwo(1.5 x parallelism)}} means that frequently, half of
the tasks get one keygroup and the other half gets two keygroups, which is very
skewed.
See section (1) in this "lessons learned" blog post.
https://engineering.contentsquare.com/2021/ten-flink-gotchas/
We can fix this by
- either setting a default maxParallelism to something pretty high (2048 for
example). The cost is that we add the default key group overhead per state
entry from one byte to two bytes.
- or we stay with some similar logic, but we instead of {{1.5 x
operatorParallelism}} we go with some higher multiplier, like {{4 x
operatorParallelism}}. The price is again that we more quickly reach the point
where we have two bytes of keygroup encoding overhead, instead of one.
Implementation wise, there is an unfortunate situation that the maxParallelism,
if not configured, is not stored anywhere in the job graph, but re-derived on
the JobManager each time it loads a JobGraph vertex (ExecutionJobVertex) which
does not have a MaxParallelism configured. This relies on the implicit contract
that this logic never changes.
Changing this logic will instantly break all jobs which have not explicitly
configured the Max Parallelism. That seems like a pretty heavy design
shortcoming, unfortunately :-(
A way to partially work around that is by moving the logic that derives the
maximum parallelism to the {{StreamGraphGenerator}}, so we never create
JobGraphs where vertices have no configured Max Parallelism (and we keep the
re-derivation logic for backwards compatibility for persisted JobGraphs).
The {{StreamExecutionEnvironment}} will need a flag to use the "old mode" to
give existing un-configured applications a way to keep restoring from old
savepoints.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
