[FLINK-29663] Further improvements of adaptive batch scheduler

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Done
• Fix Version/s: 1.17.0
• Component/s: Runtime / Coordination

Release Note

In 1.17, we further enhanced the adaptive batch scheduler, mainly including:
(1) Support for evenly distributing data to downstream tasks
(2) Remove the limitation that the decided parallelism of vertices can only be a power of 2

• Assignee: Lijie Wang
• Reporter: Lijie Wang
• Created: 17/Oct/22 11:41
• Updated: 31/Jan/23 07:38
• Resolved: 31/Jan/23 07:37

Description

In Flink 1.15, we introduced the adaptive batch scheduler to automatically decide parallelisms of job vertices for batch jobs.  In this issue, we will  further optimize it by changing the subpartition range division algorithm: change it from dividing according to the number of subpartitions(the number of subpartitions within each subpartition range is basically the same) to dividing according to the amount of data in subpartition ranges (the amount of data within each subpartition range is basically the same).

More details see https://docs.google.com/document/d/1Qyq3qDkBCUNupajVJpFTKp3fHQJtwIu9luM7T52k1Oo

 This is the umbrella ticket for the improvements.

Issue Links

• is related to: FLINK-24892 FLIP-187: Adaptive Batch Scheduler

Sub-Tasks

• Collect subpartition sizes of BLOCKING result partitions: Closed
• Support flexible subpartition range divisions: Closed
• Let adaptive batch scheduler divide subpartition range according to amount of data: Closed
• Remove the limitation that parallelism decided by adaptive batch scheduler is always a power of two: Closed
• Ignore broadcast bytes when computing parallelism and input infos for adaptive batch scheduler: Closed
• Limit the max number of subpartitons consumed by each downstream task: Closed