Stefan Richter created FLINK-5052:
-------------------------------------
Summary: Changing the maximum parallelism (number of key groups)
of a job
Key: FLINK-5052
URL: https://issues.apache.org/jira/browse/FLINK-5052
Project: Flink
Issue Type: Improvement
Components: State Backends, Checkpointing
Reporter: Stefan Richter
Through dynamic rescaling, Flink jobs can already adjust their parallelism and
each operator only has to read it's assigned key-groups.
However, the maximum parallelism is determined by the number of key-groups
(aka maxParallelism), which is currently fixed forever after the job is first
started. We could consider to relax this limitations, so that users can modify
the number of key-groups after the fact, which is useful in particular for
upscaling jobs from older Flink versions (<1.2) which must be converted with
maxparallelism == parallelism.
In the general case, changing the maxParallelism can lead to shuffling of keys
between key-groups, which means that a change in the number of key-groups can
shuffle keys between key-groups and we would have to read the complete state on
each operator instance, filtering for those keys that actually fall into the
key-groups assigned to the operator instances. While it is certainly possible
to support this, it is obviously a very expensive operation.
Fortunately, the assignment of keys to operators is currently determined as
follows:
{{operatorInstance = computeKeyGroup(key) * parallelism / maxParallelism}}
This means that we can provide more efficient support for upscaling of
maxParallelism, if {{newMaxParallelism == n * oldMaxParallelism}}. In this
case, keys are not reshuffled between key-groups, but key-groups are split by a
factor n instead. This only focus on some old key-groups when restoring
operator instances for new maxParallelism and significantly reduces the amount
of unnecessary data transfer, e.g. ~ 1/2 for increasing maxParallelism by a
factor 2, ~2/3 when increasing by a factor 3, etc.
Implementing this feature would require the following steps:
- Introduce/modify state handles with the capability to summarize
multiple logical keygroups into one mixed physical entity.
- Enhance StateAssignmentOperation so that it can deal with and
correctly assign the new/modified keyed state handles to subtasks on restoring
a checkpoint. We also need to implement how to compute the correct
super-key-group, but this is rather simple.
- Filtering out key clippings on restoring in the backends.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
