1996fanrui commented on code in PR #879:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/879#discussion_r1758222419
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -389,15 +407,61 @@ protected static int scale(
return newParallelism;
Review Comment:
We don't need to adjust parallelism via maxParallelism when source vertex
doesn't have partition info, right?
```suggestion
var adjustByMaxParallelism =
(inputShipStrategies.isEmpty() && numPartitions > 0) ||
inputShipStrategies.contains(HASH);
if (!adjustByMaxParallelism) {
return newParallelism;
```
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java:
##########
@@ -53,6 +53,9 @@ public enum ScalingMetric {
/** Job vertex max parallelism. */
MAX_PARALLELISM(false),
+
+ /** Source vertex partition count. */
+ NUM_PARTITIONS(false),
Review Comment:
IIUC, the partition is only supported for source, right?
If yes, could we rename it to `SOURCE_PARTITIONS_NUM` or
`NUM_SOURCE_PARTITIONS` ?
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -389,15 +407,61 @@ protected static int scale(
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) {
+ int numKeyGroupsOrPartitions = maxParallelism;
+ int upperBoundForAlignment;
+ if (numPartitions <= 0) {
+ upperBoundForAlignment = Math.min(maxParallelism / 2, upperBound);
Review Comment:
I'm curious why `maxParallelism / 2`?
Assuming the upstream edge of vertex is keyBy(hash):
- The maxParallelism is 100
- newParallelism is 80
We will use 80 as the result, right? If so, it will meet same issue with
source partition.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -389,15 +407,61 @@ protected static int scale(
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) {
+ int numKeyGroupsOrPartitions = maxParallelism;
+ int upperBoundForAlignment;
+ if (numPartitions <= 0) {
+ upperBoundForAlignment = Math.min(maxParallelism / 2, upperBound);
+ } else {
+ upperBoundForAlignment = Math.min(numPartitions, upperBound);
Review Comment:
Should the source vertex respect the `maxParallelism`?
If maxParallelism is 100, source has 1000 partition, and `upperBound` is
200. The new parallelism may be greater than 100, right?
IIUC, the flink job won't run when `parallelism > maxParallelism` even if
without key group.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java:
##########
@@ -46,6 +46,8 @@ public class VertexInfo {
@Setter(AccessLevel.NONE)
private int maxParallelism;
+ @Setter private int numPartitions;
Review Comment:
Same comment with the last comment.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -389,15 +407,61 @@ protected static int scale(
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) {
+ int numKeyGroupsOrPartitions = maxParallelism;
+ int upperBoundForAlignment;
+ if (numPartitions <= 0) {
+ upperBoundForAlignment = Math.min(maxParallelism / 2, upperBound);
+ } else {
+ upperBoundForAlignment = Math.min(numPartitions, upperBound);
+ numKeyGroupsOrPartitions = numPartitions;
+ }
+
+ // 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) {
return p;
}
}
+ if (numPartitions > 0) {
+
+ // When adjust the parallelism after rounding up cannot be evenly
divided by source
+ // numPartitions, Try to find the smallest parallelism that can
satisfy the current
+ // consumption rate.
+ int p = newParallelism;
+ for (; p > 0; p--) {
+ if (numPartitions / p > numPartitions / newParallelism) {
+ if (numPartitions % p != 0) {
+ p++;
+ }
+ break;
+ }
+ }
+
+ p = Math.max(p, parallelismLowerLimit);
+
+ var message =
+ String.format(
+ SCALE_LIMITED_MESSAGE_FORMAT,
+ vertex,
+ newParallelism,
+ p,
+ String.format(
+ "numPartitions :
%s，upperBoundForAlignment(maxParallelism or parallelismUpperLimit): %s, "
+ + "parallelismLowerLimit: %s.",
+ numPartitions, upperBound,
parallelismLowerLimit));
Review Comment:
Why format it twice? Could we merge `"numPartitions :
%s，upperBoundForAlignment(maxParallelism or parallelismUpperLimit): %s,
parallelismLowerLimit: %s."` into `SCALE_LIMITED_MESSAGE_FORMAT`?
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/topology/VertexInfo.java:
##########
@@ -46,6 +46,8 @@ public class VertexInfo {
@Setter(AccessLevel.NONE)
private int maxParallelism;
+ @Setter private int numPartitions;
+
private final int originalMaxParallelism;
Review Comment:
After this PR, `originalMaxParallelism` isn't needed anymore.
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java:
##########
@@ -345,15 +356,22 @@ private boolean detectIneffectiveScaleUp(
* <p>Also, in order to ensure the data is evenly spread across subtasks,
we try to adjust the
* parallelism for source and keyed vertex such that it divides the
maxParallelism without a
* remainder.
+ *
+ * <p>This method also attempts to adjust the parallelism to ensure it
aligns well with the
+ * number of partitions if a vertex has a known partition count.
Review Comment:
```suggestion
* number of source partitions if a source vertex has a known partition
count.
```
--
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]
