You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "wanglijie95 (via GitHub)" <gi...@apache.org> on 2023/02/07 12:35:48 UTC
[GitHub] [flink] wanglijie95 opened a new pull request, #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
wanglijie95 opened a new pull request, #21880:
URL: https://github.com/apache/flink/pull/21880
## What is the purpose of the change
Currently, when using the adaptive batch scheduler, the vertex parallelism decided by forward group may be larger than the global max parallelism(which is configured by option parallelism.default or execution.batch.adaptive.auto-parallelism.max-parallelism, see FLINK-30686 for details). We will fix it in this PR.
## Verifying this change
Add test `testParallelismOfForwardGroupLargerThanGlobalMaxParallelism`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (**no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (**no**)
- The serializers: (**no**)
- The runtime per-record code paths (performance sensitive): (**no**)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (**no**)
- The S3 file system connector: (**no**)
## Documentation
- Does this pull request introduce a new feature? (**no**)
- If yes, how is the feature documented? (**not applicable**)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wanglijie95 commented on pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "wanglijie95 (via GitHub)" <gi...@apache.org>.
wanglijie95 commented on PR #21880:
URL: https://github.com/apache/flink/pull/21880#issuecomment-1423870972
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wanglijie95 commented on a diff in pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "wanglijie95 (via GitHub)" <gi...@apache.org>.
wanglijie95 commented on code in PR #21880:
URL: https://github.com/apache/flink/pull/21880#discussion_r1100933333
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java:
##########
@@ -162,10 +162,7 @@ public AdaptiveBatchScheduler(
this.vertexParallelismAndInputInfosDecider =
checkNotNull(vertexParallelismAndInputInfosDecider);
- this.forwardGroupsByJobVertexId =
- ForwardGroupComputeUtil.computeForwardGroups(
- jobGraph.getVerticesSortedTopologicallyFromSources(),
- getExecutionGraph()::getJobVertex);
+ this.forwardGroupsByJobVertexId = checkNotNull(forwardGroupsByJobVertexId);
Review Comment:
Because we need to ensure that the parallelism of vertices in the forward group has been set when calling `computeVertexParallelismStoreForDynamicGraph`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21880:
URL: https://github.com/apache/flink/pull/21880#discussion_r1101002408
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java:
##########
@@ -119,43 +119,62 @@ void testAdaptiveBatchScheduler() throws Exception {
assertThat(sink.getParallelism()).isEqualTo(10);
// check aggregatedInputDataBytes of each ExecutionVertex calculated.
- checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex);
+ checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex, 26_000L);
}
@Test
void testDecideParallelismForForwardTarget() throws Exception {
- JobGraph jobGraph = createJobGraph(true);
- Iterator<JobVertex> jobVertexIterator = jobGraph.getVertices().iterator();
- JobVertex source1 = jobVertexIterator.next();
- JobVertex source2 = jobVertexIterator.next();
- JobVertex sink = jobVertexIterator.next();
+ final JobVertex source = createJobVertex("source", SOURCE_PARALLELISM_1);
+ final JobVertex map = createJobVertex("map", -1);
+ final JobVertex sink = createJobVertex("sink", -1);
- SchedulerBase scheduler = createScheduler(jobGraph);
+ map.connectNewDataSetAsInput(
+ source, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+ sink.connectNewDataSetAsInput(
+ map, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+ map.getProducedDataSets().get(0).getConsumers().get(0).setForward(true);
+
+ SchedulerBase scheduler =
+ createScheduler(
+ new JobGraph(new JobID(), "test job", source, map, sink),
+ createCustomParallelismDecider(
+ jobVertexId -> {
+ if (jobVertexId.equals(map.getID())) {
+ return 5;
+ } else {
+ return 10;
+ }
+ }),
+ 128);
final DefaultExecutionGraph graph = (DefaultExecutionGraph) scheduler.getExecutionGraph();
+ final ExecutionJobVertex mapExecutionJobVertex = graph.getJobVertex(map.getID());
final ExecutionJobVertex sinkExecutionJobVertex = graph.getJobVertex(sink.getID());
scheduler.startScheduling();
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(-1);
assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(-1);
- // trigger source1 finished.
- transitionExecutionsState(scheduler, ExecutionState.FINISHED, source1);
+ // trigger source finished.
+ transitionExecutionsState(scheduler, ExecutionState.FINISHED, source);
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(5);
assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(-1);
- // trigger source2 finished.
- transitionExecutionsState(scheduler, ExecutionState.FINISHED, source2);
- assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(SOURCE_PARALLELISM_1);
+ // trigger map finished.
+ transitionExecutionsState(scheduler, ExecutionState.FINISHED, map);
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(5);
+ assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(5);
// check that the jobGraph is updated
- assertThat(sink.getParallelism()).isEqualTo(SOURCE_PARALLELISM_1);
+ assertThat(sink.getParallelism()).isEqualTo(5);
// check aggregatedInputDataBytes of each ExecutionVertex calculated.
- checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex);
+ checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex, 13_000L);
Review Comment:
Maybe later we can improve `IntermediateResultPartition#computeNumberOfMaxPossiblePartitionConsumers` so that for a forward consumption group, the number of subpartitions can be 1.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #21880:
URL: https://github.com/apache/flink/pull/21880#issuecomment-1420716284
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "43256d0a531b26af9939d64f488a9800dc6bcd19",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "43256d0a531b26af9939d64f488a9800dc6bcd19",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 43256d0a531b26af9939d64f488a9800dc6bcd19 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21880:
URL: https://github.com/apache/flink/pull/21880#discussion_r1100950958
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java:
##########
@@ -119,43 +119,62 @@ void testAdaptiveBatchScheduler() throws Exception {
assertThat(sink.getParallelism()).isEqualTo(10);
// check aggregatedInputDataBytes of each ExecutionVertex calculated.
- checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex);
+ checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex, 26_000L);
}
@Test
void testDecideParallelismForForwardTarget() throws Exception {
- JobGraph jobGraph = createJobGraph(true);
- Iterator<JobVertex> jobVertexIterator = jobGraph.getVertices().iterator();
- JobVertex source1 = jobVertexIterator.next();
- JobVertex source2 = jobVertexIterator.next();
- JobVertex sink = jobVertexIterator.next();
+ final JobVertex source = createJobVertex("source", SOURCE_PARALLELISM_1);
+ final JobVertex map = createJobVertex("map", -1);
+ final JobVertex sink = createJobVertex("sink", -1);
- SchedulerBase scheduler = createScheduler(jobGraph);
+ map.connectNewDataSetAsInput(
+ source, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+ sink.connectNewDataSetAsInput(
+ map, DistributionPattern.POINTWISE, ResultPartitionType.BLOCKING);
+ map.getProducedDataSets().get(0).getConsumers().get(0).setForward(true);
+
+ SchedulerBase scheduler =
+ createScheduler(
+ new JobGraph(new JobID(), "test job", source, map, sink),
+ createCustomParallelismDecider(
+ jobVertexId -> {
+ if (jobVertexId.equals(map.getID())) {
+ return 5;
+ } else {
+ return 10;
+ }
+ }),
+ 128);
final DefaultExecutionGraph graph = (DefaultExecutionGraph) scheduler.getExecutionGraph();
+ final ExecutionJobVertex mapExecutionJobVertex = graph.getJobVertex(map.getID());
final ExecutionJobVertex sinkExecutionJobVertex = graph.getJobVertex(sink.getID());
scheduler.startScheduling();
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(-1);
assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(-1);
- // trigger source1 finished.
- transitionExecutionsState(scheduler, ExecutionState.FINISHED, source1);
+ // trigger source finished.
+ transitionExecutionsState(scheduler, ExecutionState.FINISHED, source);
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(5);
assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(-1);
- // trigger source2 finished.
- transitionExecutionsState(scheduler, ExecutionState.FINISHED, source2);
- assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(SOURCE_PARALLELISM_1);
+ // trigger map finished.
+ transitionExecutionsState(scheduler, ExecutionState.FINISHED, map);
+ assertThat(mapExecutionJobVertex.getParallelism()).isEqualTo(5);
+ assertThat(sinkExecutionJobVertex.getParallelism()).isEqualTo(5);
// check that the jobGraph is updated
- assertThat(sink.getParallelism()).isEqualTo(SOURCE_PARALLELISM_1);
+ assertThat(sink.getParallelism()).isEqualTo(5);
// check aggregatedInputDataBytes of each ExecutionVertex calculated.
- checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex);
+ checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex, 13_000L);
Review Comment:
This magic number is hard to understand, would you add some comments to explain it ?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchSchedulerTest.java:
##########
@@ -119,43 +119,62 @@ void testAdaptiveBatchScheduler() throws Exception {
assertThat(sink.getParallelism()).isEqualTo(10);
// check aggregatedInputDataBytes of each ExecutionVertex calculated.
- checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex);
+ checkAggregatedInputDataBytesIsCalculated(sinkExecutionJobVertex, 26_000L);
Review Comment:
This magic number is hard to understand, would you add some comments to explain it ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk closed pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk closed pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
URL: https://github.com/apache/flink/pull/21880
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21880: [FLINK-30942][runtime] Eagerly sets the parallelism of job vertices in the parallelism-decided forward groups.
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21880:
URL: https://github.com/apache/flink/pull/21880#discussion_r1098866523
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java:
##########
@@ -162,10 +162,7 @@ public AdaptiveBatchScheduler(
this.vertexParallelismAndInputInfosDecider =
checkNotNull(vertexParallelismAndInputInfosDecider);
- this.forwardGroupsByJobVertexId =
- ForwardGroupComputeUtil.computeForwardGroups(
- jobGraph.getVerticesSortedTopologicallyFromSources(),
- getExecutionGraph()::getJobVertex);
+ this.forwardGroupsByJobVertexId = checkNotNull(forwardGroupsByJobVertexId);
Review Comment:
Why not keep the forwardGroup creating in `AdaptiveBatchScheduler`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org