You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/02/14 03:17:38 UTC
[flink] branch release-1.17 updated (44e6cfb87c1 -> e1c6352d18a)
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from 44e6cfb87c1 [FLINK-30962][python] Improve the error message during launching py4j gateway server
new c20ebde320b [hotfix] Adjust the parameters of AdaptiveBatchSchedulerITCase to make tests more reasonable
new e1c6352d18a [FLINK-30917][runtime] Let adaptive batch scheduler also respect the user-configured max parallelism when deciding parallelism
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../adaptivebatch/AdaptiveBatchScheduler.java | 5 +-
...faultVertexParallelismAndInputInfosDecider.java | 146 +++++++++++++++-----
.../VertexParallelismAndInputInfosDecider.java | 6 +-
.../runtime/scheduler/DefaultSchedulerBuilder.java | 6 +-
.../adaptivebatch/AdaptiveBatchSchedulerTest.java | 66 ++++++++-
...tVertexParallelismAndInputInfosDeciderTest.java | 147 ++++++++++-----------
.../scheduling/AdaptiveBatchSchedulerITCase.java | 7 +-
7 files changed, 258 insertions(+), 125 deletions(-)
[flink] 01/02: [hotfix] Adjust the parameters of AdaptiveBatchSchedulerITCase to make tests more reasonable
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c20ebde320b91ad87f7211ee0fd73416ef2934a6
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Fri Feb 10 21:42:38 2023 +0800
[hotfix] Adjust the parameters of AdaptiveBatchSchedulerITCase to make tests more reasonable
---
.../apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
index 2dcc33585cd..cab7aa9ba74 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/AdaptiveBatchSchedulerITCase.java
@@ -84,11 +84,11 @@ class AdaptiveBatchSchedulerITCase {
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.createLocalEnvironment(configuration);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
- env.setParallelism(4);
+ env.setParallelism(8);
final DataStream<Long> source =
env.fromSequence(0, NUMBERS_TO_PRODUCE - 1)
- .setParallelism(4)
+ .setParallelism(8)
.name("source")
.slotSharingGroup("group1");
@@ -169,7 +169,8 @@ class AdaptiveBatchSchedulerITCase {
configuration.setString(RestOptions.BIND_PORT, "0");
configuration.setLong(JobManagerOptions.SLOT_REQUEST_TIMEOUT, 5000L);
configuration.setInteger(
- BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_MAX_PARALLELISM, 2);
+ BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_MAX_PARALLELISM,
+ DEFAULT_MAX_PARALLELISM);
configuration.set(
BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_AVG_DATA_VOLUME_PER_TASK,
MemorySize.parse("150kb"));
[flink] 02/02: [FLINK-30917][runtime] Let adaptive batch scheduler also respect the user-configured max parallelism when deciding parallelism
Posted by wa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e1c6352d18a19403e3eb80736c58b842de21bc88
Author: Lijie Wang <wa...@gmail.com>
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);