This is an automated email from the ASF dual-hosted git repository. wanglijie pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit b3998324b685afc779954f7e54cc0d8f281267ec Author: Lijie Wang <[email protected]> AuthorDate: Mon Feb 6 16:46:35 2023 +0800 [FLINK-30917][runtime] Let adaptive batch scheduler also respect the user-configured max parallelism when deciding parallelism This closes #21861 --- .../adaptivebatch/AdaptiveBatchScheduler.java | 5 +- ...faultVertexParallelismAndInputInfosDecider.java | 146 +++++++++++++++----- .../VertexParallelismAndInputInfosDecider.java | 6 +- .../runtime/scheduler/DefaultSchedulerBuilder.java | 6 +- .../adaptivebatch/AdaptiveBatchSchedulerTest.java | 66 ++++++++- ...tVertexParallelismAndInputInfosDeciderTest.java | 147 ++++++++++----------- 6 files changed, 254 insertions(+), 122 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java index 160f080c0fc..8abc880b2c9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java @@ -320,7 +320,10 @@ public class AdaptiveBatchScheduler extends DefaultScheduler { final ParallelismAndInputInfos parallelismAndInputInfos = vertexParallelismAndInputInfosDecider.decideParallelismAndInputInfosForVertex( - jobVertex.getJobVertexId(), inputs, parallelism); + jobVertex.getJobVertexId(), + inputs, + parallelism, + jobVertex.getMaxParallelism()); if (parallelism == ExecutionConfig.PARALLELISM_DEFAULT) { log.info( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java index 00a8eb467c5..e7326562852 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDecider.java @@ -75,54 +75,110 @@ public class DefaultVertexParallelismAndInputInfosDecider */ private static final int MAX_NUM_SUBPARTITIONS_PER_TASK_CONSUME = 32768; - private final int maxParallelism; - private final int minParallelism; + private final int globalMaxParallelism; + private final int globalMinParallelism; private final long dataVolumePerTask; - private final int defaultSourceParallelism; + private final int globalDefaultSourceParallelism; private DefaultVertexParallelismAndInputInfosDecider( - int maxParallelism, - int minParallelism, + int globalMaxParallelism, + int globalMinParallelism, MemorySize dataVolumePerTask, - int defaultSourceParallelism) { + int globalDefaultSourceParallelism) { - checkArgument(minParallelism > 0, "The minimum parallelism must be larger than 0."); + checkArgument(globalMinParallelism > 0, "The minimum parallelism must be larger than 0."); checkArgument( - maxParallelism >= minParallelism, + globalMaxParallelism >= globalMinParallelism, "Maximum parallelism should be greater than or equal to the minimum parallelism."); checkArgument( - defaultSourceParallelism > 0, + globalDefaultSourceParallelism > 0, "The default source parallelism must be larger than 0."); checkNotNull(dataVolumePerTask); - this.maxParallelism = maxParallelism; - this.minParallelism = minParallelism; + this.globalMaxParallelism = globalMaxParallelism; + this.globalMinParallelism = globalMinParallelism; this.dataVolumePerTask = dataVolumePerTask.getBytes(); - this.defaultSourceParallelism = defaultSourceParallelism; + this.globalDefaultSourceParallelism = globalDefaultSourceParallelism; } @Override public ParallelismAndInputInfos decideParallelismAndInputInfosForVertex( JobVertexID jobVertexId, List<BlockingResultInfo> consumedResults, - int initialParallelism) { + int vertexInitialParallelism, + int vertexMaxParallelism) { checkArgument( - initialParallelism == ExecutionConfig.PARALLELISM_DEFAULT - || initialParallelism > 0); + vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT + || vertexInitialParallelism > 0); + checkArgument(vertexMaxParallelism > 0 && vertexMaxParallelism >= vertexInitialParallelism); if (consumedResults.isEmpty()) { // source job vertex int parallelism = - initialParallelism > 0 ? initialParallelism : defaultSourceParallelism; + vertexInitialParallelism > 0 + ? vertexInitialParallelism + : computeSourceParallelism(jobVertexId, vertexMaxParallelism); return new ParallelismAndInputInfos(parallelism, Collections.emptyMap()); - } else if (initialParallelism == ExecutionConfig.PARALLELISM_DEFAULT - && areAllInputsAllToAll(consumedResults) - && !areAllInputsBroadcast(consumedResults)) { - return decideParallelismAndEvenlyDistributeData( - jobVertexId, consumedResults, initialParallelism); } else { - return decideParallelismAndEvenlyDistributeSubpartitions( - jobVertexId, consumedResults, initialParallelism); + int minParallelism = globalMinParallelism; + int maxParallelism = globalMaxParallelism; + + if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT + && vertexMaxParallelism < minParallelism) { + LOG.info( + "The vertex maximum parallelism {} is smaller than the global minimum parallelism {}. " + + "Use {} as the lower bound to decide parallelism of job vertex {}.", + vertexMaxParallelism, + minParallelism, + vertexMaxParallelism, + jobVertexId); + minParallelism = vertexMaxParallelism; + } + if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT + && vertexMaxParallelism < maxParallelism) { + LOG.info( + "The vertex maximum parallelism {} is smaller than the global maximum parallelism {}. " + + "Use {} as the upper bound to decide parallelism of job vertex {}.", + vertexMaxParallelism, + maxParallelism, + vertexMaxParallelism, + jobVertexId); + maxParallelism = vertexMaxParallelism; + } + checkState(maxParallelism >= minParallelism); + + if (vertexInitialParallelism == ExecutionConfig.PARALLELISM_DEFAULT + && areAllInputsAllToAll(consumedResults) + && !areAllInputsBroadcast(consumedResults)) { + return decideParallelismAndEvenlyDistributeData( + jobVertexId, + consumedResults, + vertexInitialParallelism, + minParallelism, + maxParallelism); + } else { + return decideParallelismAndEvenlyDistributeSubpartitions( + jobVertexId, + consumedResults, + vertexInitialParallelism, + minParallelism, + maxParallelism); + } + } + } + + private int computeSourceParallelism(JobVertexID jobVertexId, int maxParallelism) { + if (globalDefaultSourceParallelism > maxParallelism) { + LOG.info( + "The global default source parallelism {} is larger than the maximum parallelism {}. " + + "Use {} as the parallelism of source job vertex {}.", + globalDefaultSourceParallelism, + maxParallelism, + maxParallelism, + jobVertexId); + return maxParallelism; + } else { + return globalDefaultSourceParallelism; } } @@ -142,24 +198,33 @@ public class DefaultVertexParallelismAndInputInfosDecider * @param jobVertexId The job vertex id * @param consumedResults The information of consumed blocking results * @param initialParallelism The initial parallelism of the job vertex + * @param minParallelism the min parallelism + * @param maxParallelism the max parallelism * @return the parallelism and vertex input infos */ private ParallelismAndInputInfos decideParallelismAndEvenlyDistributeSubpartitions( JobVertexID jobVertexId, List<BlockingResultInfo> consumedResults, - int initialParallelism) { + int initialParallelism, + int minParallelism, + int maxParallelism) { checkArgument(!consumedResults.isEmpty()); int parallelism = initialParallelism > 0 ? initialParallelism - : decideParallelism(jobVertexId, consumedResults); + : decideParallelism( + jobVertexId, consumedResults, minParallelism, maxParallelism); return new ParallelismAndInputInfos( parallelism, VertexInputInfoComputationUtils.computeVertexInputInfos( parallelism, consumedResults, true)); } - int decideParallelism(JobVertexID jobVertexId, List<BlockingResultInfo> consumedResults) { + int decideParallelism( + JobVertexID jobVertexId, + List<BlockingResultInfo> consumedResults, + int minParallelism, + int maxParallelism) { checkArgument(!consumedResults.isEmpty()); // Considering that the sizes of broadcast results are usually very small, we compute the @@ -219,12 +284,16 @@ public class DefaultVertexParallelismAndInputInfosDecider * @param jobVertexId The job vertex id * @param consumedResults The information of consumed blocking results * @param initialParallelism The initial parallelism of the job vertex + * @param minParallelism the min parallelism + * @param maxParallelism the max parallelism * @return the parallelism and vertex input infos */ private ParallelismAndInputInfos decideParallelismAndEvenlyDistributeData( JobVertexID jobVertexId, List<BlockingResultInfo> consumedResults, - int initialParallelism) { + int initialParallelism, + int minParallelism, + int maxParallelism) { checkArgument(initialParallelism == ExecutionConfig.PARALLELISM_DEFAULT); checkArgument(!consumedResults.isEmpty()); consumedResults.forEach(resultInfo -> checkState(!resultInfo.isPointwise())); @@ -252,11 +321,13 @@ public class DefaultVertexParallelismAndInputInfosDecider computeSubpartitionRanges(bytesBySubpartition, dataVolumePerTask, maxRangeSize); // if the parallelism is not legal, adjust to a legal parallelism - if (!isLegalParallelism(subpartitionRanges.size())) { + if (!isLegalParallelism(subpartitionRanges.size(), minParallelism, maxParallelism)) { Optional<List<IndexRange>> adjustedSubpartitionRanges = adjustToClosestLegalParallelism( dataVolumePerTask, subpartitionRanges.size(), + minParallelism, + maxParallelism, Arrays.stream(bytesBySubpartition).min().getAsLong(), Arrays.stream(bytesBySubpartition).sum(), limit -> computeParallelism(bytesBySubpartition, limit, maxRangeSize), @@ -270,16 +341,21 @@ public class DefaultVertexParallelismAndInputInfosDecider + "Fall back to compute a parallelism that can evenly distribute subpartitions.", jobVertexId); return decideParallelismAndEvenlyDistributeSubpartitions( - jobVertexId, consumedResults, initialParallelism); + jobVertexId, + consumedResults, + initialParallelism, + minParallelism, + maxParallelism); } subpartitionRanges = adjustedSubpartitionRanges.get(); } - checkState(isLegalParallelism(subpartitionRanges.size())); + checkState(isLegalParallelism(subpartitionRanges.size(), minParallelism, maxParallelism)); return createParallelismAndInputInfos(consumedResults, subpartitionRanges); } - private boolean isLegalParallelism(int parallelism) { + private static boolean isLegalParallelism( + int parallelism, int minParallelism, int maxParallelism) { return parallelism >= minParallelism && parallelism <= maxParallelism; } @@ -303,6 +379,8 @@ public class DefaultVertexParallelismAndInputInfosDecider * * @param currentDataVolumeLimit current data volume limit * @param currentParallelism current parallelism + * @param minParallelism the min parallelism + * @param maxParallelism the max parallelism * @param minLimit the minimum data volume limit * @param maxLimit the maximum data volume limit * @param parallelismComputer a function to compute the parallelism according to the data volume @@ -312,9 +390,11 @@ public class DefaultVertexParallelismAndInputInfosDecider * @return the computed subpartition ranges or {@link Optional#empty()} if we can't find any * legal parallelism */ - private Optional<List<IndexRange>> adjustToClosestLegalParallelism( + private static Optional<List<IndexRange>> adjustToClosestLegalParallelism( long currentDataVolumeLimit, int currentParallelism, + int minParallelism, + int maxParallelism, long minLimit, long maxLimit, Function<Long, Integer> parallelismComputer, @@ -355,7 +435,7 @@ public class DefaultVertexParallelismAndInputInfosDecider } int adjustedParallelism = parallelismComputer.apply(adjustedDataVolumeLimit); - if (isLegalParallelism(adjustedParallelism)) { + if (isLegalParallelism(adjustedParallelism, minParallelism, maxParallelism)) { return Optional.of(subpartitionRangesComputer.apply(adjustedDataVolumeLimit)); } else { return Optional.empty(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismAndInputInfosDecider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismAndInputInfosDecider.java index 28cc8f13336..b2144a10de7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismAndInputInfosDecider.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/VertexParallelismAndInputInfosDecider.java @@ -37,14 +37,16 @@ public interface VertexParallelismAndInputInfosDecider { * * @param jobVertexId The job vertex id * @param consumedResults The information of consumed blocking results - * @param initialParallelism The initial parallelism of the job vertex. If it's a positive + * @param vertexInitialParallelism The initial parallelism of the job vertex. If it's a positive * number, it will be respected. If it's not set(equals to {@link * ExecutionConfig#PARALLELISM_DEFAULT}), a parallelism will be automatically decided for * the vertex. + * @param vertexMaxParallelism The max parallelism of the job vertex. * @return the parallelism and vertex input infos. */ ParallelismAndInputInfos decideParallelismAndInputInfosForVertex( JobVertexID jobVertexId, List<BlockingResultInfo> consumedResults, - int initialParallelism); + int vertexInitialParallelism, + int vertexMaxParallelism); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBuilder.java index 374e10ea8b6..d273c437920 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBuilder.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSchedulerBuilder.java @@ -402,10 +402,10 @@ public class DefaultSchedulerBuilder { public static VertexParallelismAndInputInfosDecider createCustomParallelismDecider( Function<JobVertexID, Integer> parallelismFunction) { - return (jobVertexId, consumedResults, initialParallelism) -> { + return (jobVertexId, consumedResults, vertexInitialParallelism, ignored) -> { int parallelism = - initialParallelism > 0 - ? initialParallelism + vertexInitialParallelism > 0 + ? vertexInitialParallelism : parallelismFunction.apply(jobVertexId); return new ParallelismAndInputInfos( parallelism, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java index 559028683e7..2f4acebdbaa 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java @@ -67,7 +67,7 @@ import java.util.stream.LongStream; import static org.apache.flink.runtime.scheduler.DefaultSchedulerBuilder.createCustomParallelismDecider; import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createFailedTaskExecutionState; import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createFinishedTaskExecutionState; -import static org.apache.flink.runtime.scheduler.adaptivebatch.DefaultVertexParallelismAndInputInfosDeciderTest.createDefaultVertexParallelismAndInputInfosDecider; +import static org.apache.flink.runtime.scheduler.adaptivebatch.DefaultVertexParallelismAndInputInfosDeciderTest.createDecider; import static org.apache.flink.shaded.guava30.com.google.common.collect.Iterables.getOnlyElement; import static org.assertj.core.api.Assertions.assertThat; @@ -266,8 +266,7 @@ class AdaptiveBatchSchedulerTest { SchedulerBase scheduler = createScheduler( new JobGraph(new JobID(), "test job", source, sink), - createDefaultVertexParallelismAndInputInfosDecider( - 1, 16, 4 * SUBPARTITION_BYTES), + createDecider(1, 16, 4 * SUBPARTITION_BYTES), 16); final DefaultExecutionGraph graph = (DefaultExecutionGraph) scheduler.getExecutionGraph(); @@ -298,6 +297,67 @@ class AdaptiveBatchSchedulerTest { assertThat(sinkExecutionJobVertex.isInitialized()).isTrue(); } + @Test + void testUserConfiguredMaxParallelismIsLargerThanGlobalMaxParallelism() throws Exception { + testUserConfiguredMaxParallelism(1, 32, 128, 1L, 32); + } + + @Test + void testUserConfiguredMaxParallelismIsSmallerThanGlobalMaxParallelism() throws Exception { + testUserConfiguredMaxParallelism(1, 128, 32, 1L, 32); + } + + @Test + void testUserConfiguredMaxParallelismIsSmallerThanGlobalMinParallelism() throws Exception { + testUserConfiguredMaxParallelism(16, 128, 8, 4 * SUBPARTITION_BYTES, 8); + } + + @Test + void testUserConfiguredMaxParallelismIsSmallerThanGlobalDefaultSourceParallelism() + throws Exception { + final JobVertex source = createJobVertex("source", -1); + source.setMaxParallelism(8); + + SchedulerBase scheduler = + createScheduler( + new JobGraph(new JobID(), "test job", source), + createDecider(1, 128, 1L, 32), + 128); + + scheduler.startScheduling(); + + // check source's parallelism + assertThat(source.getParallelism()).isEqualTo(8); + } + + void testUserConfiguredMaxParallelism( + int globalMinParallelism, + int globalMaxParallelism, + int userConfiguredMaxParallelism, + long dataVolumePerTask, + int expectedParallelism) + throws Exception { + final JobVertex source = createJobVertex("source", 8); + final JobVertex sink = createJobVertex("sink", -1); + sink.setMaxParallelism(userConfiguredMaxParallelism); + + sink.connectNewDataSetAsInput( + source, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING); + + SchedulerBase scheduler = + createScheduler( + new JobGraph(new JobID(), "test job", source, sink), + createDecider( + globalMinParallelism, globalMaxParallelism, dataVolumePerTask), + globalMaxParallelism); + + scheduler.startScheduling(); + transitionExecutionsState(scheduler, ExecutionState.FINISHED, source); + + // check sink's parallelism + assertThat(sink.getParallelism()).isEqualTo(expectedParallelism); + } + private BlockingResultInfo getBlockingResultInfo( AdaptiveBatchScheduler scheduler, JobVertex jobVertex) { return scheduler.getBlockingResultInfo( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDeciderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDeciderTest.java index 90f2287c050..2596fe23329 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDeciderTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/DefaultVertexParallelismAndInputInfosDeciderTest.java @@ -59,83 +59,61 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testDecideParallelism() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); - BlockingResultInfo resultInfo1 = createFromBroadcastResult(BYTE_256_MB); BlockingResultInfo resultInfo2 = createFromNonBroadcastResult(BYTE_256_MB + BYTE_8_GB); int parallelism = - decider.decideParallelism( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2)); + createDeciderAndDecideParallelism(Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelism).isEqualTo(9); } @Test void testInitiallyNormalizedParallelismIsLargerThanMaxParallelism() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); - BlockingResultInfo resultInfo1 = createFromBroadcastResult(BYTE_256_MB); BlockingResultInfo resultInfo2 = createFromNonBroadcastResult(BYTE_8_GB + BYTE_1_TB); int parallelism = - decider.decideParallelism( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2)); + createDeciderAndDecideParallelism(Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelism).isEqualTo(MAX_PARALLELISM); } @Test void testInitiallyNormalizedParallelismIsSmallerThanMinParallelism() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); - BlockingResultInfo resultInfo1 = createFromBroadcastResult(BYTE_256_MB); BlockingResultInfo resultInfo2 = createFromNonBroadcastResult(BYTE_512_MB); int parallelism = - decider.decideParallelism( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2)); + createDeciderAndDecideParallelism(Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelism).isEqualTo(MIN_PARALLELISM); } @Test void testNonBroadcastBytesCanNotDividedEvenly() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); - BlockingResultInfo resultInfo1 = createFromBroadcastResult(BYTE_512_MB); BlockingResultInfo resultInfo2 = createFromNonBroadcastResult(BYTE_256_MB + BYTE_8_GB); int parallelism = - decider.decideParallelism( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2)); + createDeciderAndDecideParallelism(Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelism).isEqualTo(9); } @Test void testDecideParallelismWithMaxSubpartitionLimitation() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 100, BYTE_256_MB); - BlockingResultInfo resultInfo1 = new TestingBlockingResultInfo(false, 1L, 1024, 1024); BlockingResultInfo resultInfo2 = new TestingBlockingResultInfo(false, 1L, 512, 512); int parallelism = - decider.decideParallelism( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2)); + createDeciderAndDecideParallelism( + 1, 100, BYTE_256_MB, Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelism).isEqualTo(32); } @Test void testAllEdgesAllToAll() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 10, 60L); - AllToAllBlockingResultInfo resultInfo1 = createAllToAllBlockingResultInfo( new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L}); @@ -143,8 +121,8 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { createAllToAllBlockingResultInfo( new long[] {8L, 12L, 21L, 9L, 13L, 7L, 19L, 13L, 14L, 5L}); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 10, 60L, Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(5); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(2); @@ -166,15 +144,12 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testAllEdgesAllToAllAndDecidedParallelismIsMaxParallelism() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 2, 10L); - AllToAllBlockingResultInfo resultInfo = createAllToAllBlockingResultInfo( new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L}); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Collections.singletonList(resultInfo), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 2, 10L, Collections.singletonList(resultInfo)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(2); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(1); @@ -186,15 +161,12 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testAllEdgesAllToAllAndDecidedParallelismIsMinParallelism() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(4, 10, 1000L); - AllToAllBlockingResultInfo resultInfo = createAllToAllBlockingResultInfo( new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L}); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Collections.singletonList(resultInfo), -1); + createDeciderAndDecideParallelismAndInputInfos( + 4, 10, 1000L, Collections.singletonList(resultInfo)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(4); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(1); @@ -210,15 +182,12 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testFallBackToEvenlyDistributeSubpartitions() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(8, 8, 10L); - AllToAllBlockingResultInfo resultInfo = createAllToAllBlockingResultInfo( new long[] {10L, 1L, 10L, 1L, 10L, 1L, 10L, 1L, 10L, 1L}); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Collections.singletonList(resultInfo), -1); + createDeciderAndDecideParallelismAndInputInfos( + 8, 8, 10L, Collections.singletonList(resultInfo)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(8); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(1); @@ -238,9 +207,6 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testAllEdgesAllToAllAndOneIsBroadcast() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 10, 60L); - AllToAllBlockingResultInfo resultInfo1 = createAllToAllBlockingResultInfo( new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L}); @@ -248,8 +214,8 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { createAllToAllBlockingResultInfo(new long[] {10L}, true); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 10, 60L, Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(3); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(2); @@ -264,16 +230,13 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testAllEdgesBroadcast() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 10, 60L); - AllToAllBlockingResultInfo resultInfo1 = createAllToAllBlockingResultInfo(new long[] {10L}, true); AllToAllBlockingResultInfo resultInfo2 = createAllToAllBlockingResultInfo(new long[] {10L}, true); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 10, 60L, Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(1); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(2); @@ -288,9 +251,6 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testHavePointwiseEdges() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 10, 60L); - AllToAllBlockingResultInfo resultInfo1 = createAllToAllBlockingResultInfo( new long[] {10L, 15L, 13L, 12L, 1L, 10L, 8L, 20L, 12L, 17L}); @@ -298,8 +258,8 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { createPointwiseBlockingResultInfo( new long[] {8L, 12L, 21L, 9L, 13L}, new long[] {7L, 19L, 13L, 14L, 5L}); ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Arrays.asList(resultInfo1, resultInfo2), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 10, 60L, Arrays.asList(resultInfo1, resultInfo2)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(4); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(2); @@ -328,7 +288,7 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testParallelismAlreadyDecided() { final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); + createDecider(MIN_PARALLELISM, MAX_PARALLELISM, DATA_VOLUME_PER_TASK); AllToAllBlockingResultInfo allToAllBlockingResultInfo = createAllToAllBlockingResultInfo( @@ -337,7 +297,8 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { decider.decideParallelismAndInputInfosForVertex( new JobVertexID(), Collections.singletonList(allToAllBlockingResultInfo), - 3); + 3, + MAX_PARALLELISM); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(3); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).hasSize(1); @@ -350,12 +311,12 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testSourceJobVertex() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(); - ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Collections.emptyList(), -1); + createDeciderAndDecideParallelismAndInputInfos( + MIN_PARALLELISM, + MAX_PARALLELISM, + DATA_VOLUME_PER_TASK, + Collections.emptyList()); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(DEFAULT_SOURCE_PARALLELISM); assertThat(parallelismAndInputInfos.getJobVertexInputInfos()).isEmpty(); @@ -363,9 +324,6 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { @Test void testEvenlyDistributeDataWithMaxSubpartitionLimitation() { - final DefaultVertexParallelismAndInputInfosDecider decider = - createDefaultVertexParallelismAndInputInfosDecider(1, 100, BYTE_256_MB); - long[] subpartitionBytes = new long[1024]; Arrays.fill(subpartitionBytes, 1L); AllToAllBlockingResultInfo resultInfo = @@ -375,8 +333,8 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { } ParallelismAndInputInfos parallelismAndInputInfos = - decider.decideParallelismAndInputInfosForVertex( - new JobVertexID(), Collections.singletonList(resultInfo), -1); + createDeciderAndDecideParallelismAndInputInfos( + 1, 100, BYTE_256_MB, Collections.singletonList(resultInfo)); assertThat(parallelismAndInputInfos.getParallelism()).isEqualTo(32); List<IndexRange> subpartitionRanges = new ArrayList<>(); @@ -424,15 +382,17 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { .containsExactlyInAnyOrderElementsOf(executionVertexInputInfos); } - private static DefaultVertexParallelismAndInputInfosDecider - createDefaultVertexParallelismAndInputInfosDecider() { - return createDefaultVertexParallelismAndInputInfosDecider( - MIN_PARALLELISM, MAX_PARALLELISM, DATA_VOLUME_PER_TASK); + static DefaultVertexParallelismAndInputInfosDecider createDecider( + int minParallelism, int maxParallelism, long dataVolumePerTask) { + return createDecider( + minParallelism, maxParallelism, dataVolumePerTask, DEFAULT_SOURCE_PARALLELISM); } - static DefaultVertexParallelismAndInputInfosDecider - createDefaultVertexParallelismAndInputInfosDecider( - int minParallelism, int maxParallelism, long dataVolumePerTask) { + static DefaultVertexParallelismAndInputInfosDecider createDecider( + int minParallelism, + int maxParallelism, + long dataVolumePerTask, + int defaultSourceParallelism) { Configuration configuration = new Configuration(); configuration.setInteger( @@ -442,11 +402,38 @@ class DefaultVertexParallelismAndInputInfosDeciderTest { new MemorySize(dataVolumePerTask)); configuration.setInteger( BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_DEFAULT_SOURCE_PARALLELISM, - DEFAULT_SOURCE_PARALLELISM); + defaultSourceParallelism); return DefaultVertexParallelismAndInputInfosDecider.from(maxParallelism, configuration); } + private static int createDeciderAndDecideParallelism(List<BlockingResultInfo> consumedResults) { + return createDeciderAndDecideParallelism( + MIN_PARALLELISM, MAX_PARALLELISM, DATA_VOLUME_PER_TASK, consumedResults); + } + + private static int createDeciderAndDecideParallelism( + int minParallelism, + int maxParallelism, + long dataVolumePerTask, + List<BlockingResultInfo> consumedResults) { + final DefaultVertexParallelismAndInputInfosDecider decider = + createDecider(minParallelism, maxParallelism, dataVolumePerTask); + return decider.decideParallelism( + new JobVertexID(), consumedResults, minParallelism, maxParallelism); + } + + private static ParallelismAndInputInfos createDeciderAndDecideParallelismAndInputInfos( + int minParallelism, + int maxParallelism, + long dataVolumePerTask, + List<BlockingResultInfo> consumedResults) { + final DefaultVertexParallelismAndInputInfosDecider decider = + createDecider(minParallelism, maxParallelism, dataVolumePerTask); + return decider.decideParallelismAndInputInfosForVertex( + new JobVertexID(), consumedResults, -1, maxParallelism); + } + private AllToAllBlockingResultInfo createAllToAllBlockingResultInfo( long[] aggregatedSubpartitionBytes) { return createAllToAllBlockingResultInfo(aggregatedSubpartitionBytes, false);
