zhijiangW commented on a change in pull request #7631: [FLINK-11391][shuffle]
Introduce PartitionShuffleDescriptor and ShuffleDeploymentDescriptor
URL: https://github.com/apache/flink/pull/7631#discussion_r255814019
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
##########
@@ -813,24 +815,27 @@ TaskDeploymentDescriptor createDeploymentDescriptor(
boolean lazyScheduling =
getExecutionGraph().getScheduleMode().allowLazyDeployment();
for (IntermediateResultPartition partition :
resultPartitions.values()) {
-
List<List<ExecutionEdge>> consumers =
partition.getConsumers();
-
+ int maxParallelism;
if (consumers.isEmpty()) {
//TODO this case only exists for test,
currently there has to be exactly one consumer in real jobs!
-
producedPartitions.add(ResultPartitionDeploymentDescriptor.from(
- partition,
-
KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM,
- lazyScheduling));
+ maxParallelism =
KeyGroupRangeAssignment.UPPER_BOUND_MAX_PARALLELISM;
} else {
Preconditions.checkState(1 == consumers.size(),
- "Only one consumer supported in
the current implementation! Found: " + consumers.size());
+ "Only one consumer supported in the
current implementation! Found: " + consumers.size());
List<ExecutionEdge> consumer = consumers.get(0);
ExecutionJobVertex vertex =
consumer.get(0).getTarget().getJobVertex();
- int maxParallelism = vertex.getMaxParallelism();
-
producedPartitions.add(ResultPartitionDeploymentDescriptor.from(partition,
maxParallelism, lazyScheduling));
+ maxParallelism = vertex.getMaxParallelism();
}
+
+ PartitionShuffleDescriptor psd =
PartitionShuffleDescriptor.from(targetSlot, executionId, partition,
maxParallelism);
+
+
producedPartitions.add(ResultPartitionDeploymentDescriptor.fromShuffleDescriptor(psd));
+
getCurrentExecutionAttempt().cachePartitionShuffleDescriptor(partition.getIntermediateResult().getId(),
psd);
Review comment:
From functional aspect, caching the `TaskDeploymentDescriptor` might also
make sense. But I have other concerns:
1. The structure of TDD is complicated and would take more memory if caching
completely, such as unnecessary fields `serializedJobInformation`,
`serializedTaskInformation`,etc.
2. We might need adjust the current collection structure of
`producedPartitions`, `inputGates` in TDD to map structure in order to find
required PSD, SDD directly for other usages.
3. If replacing the current three descriptors caches, we might not need the
class of `PartialInputChannelDeploymentDescriptor` any more if I understand
correctly. But I wonder there exists such scenarios that during deploying
consumer execution, only some input channel descriptors are unknown. During
sending partition infos we only want to send these unknown infos, so how can we
distinguish them from all the cached producer's TDD? In other words, the
current cached `partialInputChannelDeploymentDescriptors` might be only a sub
collection of all cached TDDs on producer side.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
