[FLINK-36576][runtime] Improving amount-based data balancing distribu…

…tion algorithm for DefaultVertexParallelismAndInputInfosDecider
apache · Jan 11, 2025 · 7f1c201 · 7f1c201
1 parent 4860f38
commit 7f1c201
Show file tree

Hide file tree

Showing 17 changed files with 3,204 additions and 441 deletions.
diff --git a/...rc/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java b/...rc/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java
@@ -104,7 +104,7 @@ public static Map<IntermediateDataSetID, JobVertexInputInfo> computeVertexInputI
      * @param isDynamicGraph whether is dynamic graph
      * @return the computed {@link JobVertexInputInfo}
      */
-    static JobVertexInputInfo computeVertexInputInfoForPointwise(
+    public static JobVertexInputInfo computeVertexInputInfoForPointwise(
             int sourceCount,
             int targetCount,
             Function<Integer, Integer> numOfSubpartitionsRetriever,

diff --git a/.../java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java b/.../java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerFactory.java
@@ -199,6 +199,17 @@ public SchedulerNG createInstance(
                 new ScheduledExecutorServiceAdapter(futureExecutor),
                 DefaultVertexParallelismAndInputInfosDecider.from(
                         getDefaultMaxParallelism(jobMasterConfiguration, executionConfig),
+                        executionPlan
+                                .getJobConfiguration()
+                                .get(
+                                        BatchExecutionOptionsInternal
+                                                .ADAPTIVE_SKEWED_OPTIMIZATION_SKEWED_FACTOR),
+                        executionPlan
+                                .getJobConfiguration()
+                                .get(
+                                        BatchExecutionOptionsInternal
+                                                .ADAPTIVE_SKEWED_OPTIMIZATION_SKEWED_THRESHOLD)
+                                .getBytes(),
                         jobMasterConfiguration),
                 jobRecoveryHandler);
     }

diff --git a/.../java/org/apache/flink/runtime/scheduler/adaptivebatch/BatchExecutionOptionsInternal.java b/.../java/org/apache/flink/runtime/scheduler/adaptivebatch/BatchExecutionOptionsInternal.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler.adaptivebatch;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.MemorySize;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/** Internal configuration options for the batch job execution. */
+@Internal
+public class BatchExecutionOptionsInternal {
+    public static final ConfigOption<MemorySize> ADAPTIVE_SKEWED_OPTIMIZATION_SKEWED_THRESHOLD =
+            key("$internal.execution.batch.adaptive.skewed-optimization.skewed-threshold")
+                    .memoryType()
+                    .defaultValue(MemorySize.ofMebiBytes(256))
+                    .withDescription(
+                            "Flink will automatically reduce the ratio of the maximum to median concurrent task "
+                                    + "processing data volume to below the skewed-factor and will also achieve "
+                                    + "a more balanced data distribution, unless the maximum value is below the "
+                                    + "skewed-threshold.");
+
+    public static final ConfigOption<Double> ADAPTIVE_SKEWED_OPTIMIZATION_SKEWED_FACTOR =
+            key("$internal.execution.batch.adaptive.skewed-optimization.skewed-factor")
+                    .doubleType()
+                    .defaultValue(4.0)
+                    .withDescription(
+                            "When the maximum data volume processed by a concurrent task is greater than the "
+                                    + "skewed-threshold, Flink can automatically reduce the ratio of the maximum "
+                                    + "data volume processed by a concurrent task to the median to less than the "
+                                    + "skewed-factor and will also achieve a more balanced data distribution.");
+}
diff --git a/.../src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/BisectionSearchUtils.java b/.../src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/BisectionSearchUtils.java
@@ -22,7 +22,7 @@
 import java.util.function.Function;
 
 /** Utility class for bisection search. */
-class BisectionSearchUtils {
+public class BisectionSearchUtils {
 
     public static long findMinLegalValue(
             Function<Long, Boolean> legalChecker, long low, long high) {