1996fanrui commented on code in PR #879:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/879#discussion_r1764471043
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -383,23 +403,68 @@ protected static int scale(
// Apply min/max parallelism
newParallelism = Math.min(Math.max(parallelismLowerLimit,
newParallelism), upperBound);
- var adjustByMaxParallelism =
- inputShipStrategies.isEmpty() ||
inputShipStrategies.contains(HASH);
- if (!adjustByMaxParallelism) {
+ var adjustByMaxParallelismOrPartitions =
+ numSourcePartitions > 0 || inputShipStrategies.contains(HASH);
+ if (!adjustByMaxParallelismOrPartitions) {
return newParallelism;
}
- // When the shuffle type of vertex inputs contains keyBy or vertex is
a source, we try to
- // adjust the parallelism such that it divides the maxParallelism
without a remainder
- // => data is evenly spread across subtasks
- for (int p = newParallelism; p <= maxParallelism / 2 && p <=
upperBound; p++) {
- if (maxParallelism % p == 0) {
+ final int numKeyGroupsOrPartitions;
+ final int upperBoundForAlignment;
+ if (numSourcePartitions <= 0) {
+ numKeyGroupsOrPartitions = maxParallelism;
+ upperBoundForAlignment =
+ Math.min(
+ // Optimize the case where newParallelism <=
maxParallelism / 2
+ newParallelism > maxParallelism / 2
+ ? maxParallelism
+ : maxParallelism / 2,
+ upperBound);
+ } else {
+ numKeyGroupsOrPartitions = numSourcePartitions;
+ upperBoundForAlignment = Math.min(numSourcePartitions, upperBound);
+ }
+
+ // When the shuffle type of vertex inputs contains keyBy or vertex is
a source,
+ // we try to adjust the parallelism such that it divides
+ // the adjustableMaxParallelism without a remainder => data is evenly
spread across subtasks
+ for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
+ if (numKeyGroupsOrPartitions % p == 0) {
Review Comment:
About this comment
https://github.com/apache/flink-kubernetes-operator/pull/879#discussion_r1764419949,
I'm thinking whether the following change is more reasonable?
Note: `numKeyGroupsOrPartitions / p` means how many source partitions or key
groups every subtask consume.
```suggestion
if (numKeyGroupsOrPartitions % p == 0 ||
numKeyGroupsOrPartitions / p < numKeyGroupsOrPartitions / newParallelism) {
```
For example: maxParallelism is 200, and new parallelism is 60. (Some
subtasks consume 4 keyGroups, the rest of subtask consume 3 keyGroups)
- The final parallelism is 100 based on the main branch code due to we only
return p when `maxParallelism % p == 0`.
- But I think 67 is more reasonable here. (One subtask consumes 2 key
groups. The remaining 66 subtasks, each subtask consumes 3 key groups.)
@mxm @gyfora , WDYT?
Also, it's a bit beyond the scope of this PR. I could file a separate PR if
you think it makes sense. Of course, it's acceptable to be done at this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
