taegeonum commented on a change in pull request #128: [NEMO-235] Refactor
TaskExecutor's data processing logic
URL: https://github.com/apache/incubator-nemo/pull/128#discussion_r228027440
##########
File path:
runtime/executor/src/main/java/org/apache/nemo/runtime/executor/task/TaskExecutor.java
##########
@@ -149,38 +151,37 @@ public TaskExecutor(final Task task,
final List<DataFetcher> nonBroadcastDataFetcherList = new ArrayList<>();
final Map<String, VertexHarness> vertexIdToHarness = new HashMap<>();
reverseTopologicallySorted.forEach(irVertex -> {
- final List<VertexHarness> children = getChildrenHarnesses(irVertex,
irVertexDag, vertexIdToHarness);
final Optional<Readable> sourceReader = getSourceVertexReader(irVertex,
task.getIrVertexIdToReadable());
if (sourceReader.isPresent() != irVertex instanceof SourceVertex) {
throw new IllegalStateException(irVertex.toString());
}
- // Prepare data WRITE
- // Child-task writes
- final Map<String, String> additionalOutputMap =
- getAdditionalOutputMap(irVertex, task.getTaskOutgoingEdges(),
irVertexDag);
-
- final List<Boolean> isToAdditionalTagOutputs = children.stream()
- .map(harness -> harness.getIRVertex().getId())
- .map(additionalOutputMap::containsValue)
- .collect(Collectors.toList());
-
- // Handle writes
- // Main output children task writes
- final List<OutputWriter> mainChildrenTaskWriters =
getMainChildrenTaskWriters(
- irVertex, task.getTaskOutgoingEdges(), dataTransferFactory,
additionalOutputMap);
- final Map<String, OutputWriter> additionalChildrenTaskWriters =
getAdditionalChildrenTaskWriters(
- irVertex, task.getTaskOutgoingEdges(), dataTransferFactory,
additionalOutputMap);
- // Intra-task writes
- final List<String> additionalOutputVertices = new
ArrayList<>(additionalOutputMap.values());
- final Set<String> mainChildren =
- getMainOutputVertices(irVertex, irVertexDag,
task.getTaskOutgoingEdges(), additionalOutputVertices);
- final OutputCollectorImpl oci = new OutputCollectorImpl(mainChildren,
additionalOutputMap);
+ // Additional outputs
+ final Map<String, Set<OperatorVertex>> internalAdditionalOutputMap =
+ getInternalAdditionalOutputMap(irVertex, irVertexDag);
+ final Map<String, Set<OutputWriter>> externalAdditionalOutputMap =
+ getExternalAdditionalOutputMap(irVertex, task.getTaskOutgoingEdges(),
dataTransferFactory);
+
+ // Main outputs
+ final Set<OperatorVertex> internalMainOutputs =
getInternalMainOutputs(irVertex, irVertexDag);
+ final Set<OutputWriter> externalMainOutputs =
+ getExternalMainOutputs(irVertex, task.getTaskOutgoingEdges(),
dataTransferFactory);
+
+ OutputCollector outputCollector = new OutputCollectorImpl(
+ irVertex, internalMainOutputs, internalAdditionalOutputMap,
+ externalMainOutputs, externalAdditionalOutputMap);
+
+ if (irVertex instanceof OperatorVertex
+ && ((OperatorVertex) irVertex).getTransform() instanceof
AggregateMetricTransform) {
+ outputCollector = new DynOptDataOutputCollector(
+ irVertex, persistentConnectionToMasterMap, this);
+ }
Review comment:
addressed
----------------------------------------------------------------
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
