beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [20/36] beam git commit: mr-runner: fix the bug that steps are attached multiple times in diamond shaped DAG.
Date Thu, 07 Sep 2017 18:39:29 GMT
mr-runner: fix the bug that steps are attached multiple times in diamond shaped DAG.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/4e7062cd
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/4e7062cd
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/4e7062cd

Branch: refs/heads/mr-runner
Commit: 4e7062cd9de6ae3f0616033823fd995eb10a3744
Parents: 6c2390a
Author: Pei He <pei@apache.org>
Authored: Wed Aug 30 19:16:06 2017 +0800
Committer: Pei He <pei@apache.org>
Committed: Thu Aug 31 14:13:50 2017 +0800

----------------------------------------------------------------------
 .../runners/mapreduce/translation/JobPrototype.java   | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/4e7062cd/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/translation/JobPrototype.java
----------------------------------------------------------------------
diff --git a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/translation/JobPrototype.java
b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/translation/JobPrototype.java
index 39487fd..a0c6626 100644
--- a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/translation/JobPrototype.java
+++ b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/translation/JobPrototype.java
@@ -24,10 +24,12 @@ import com.google.common.base.Function;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import org.apache.beam.runners.mapreduce.MapReducePipelineOptions;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.KvCoder;
@@ -139,7 +141,7 @@ public class JobPrototype {
 
       // Setup BeamReducer
       Graphs.Step reducerStartStep = gabwStep;
-      chainOperations(reducerStartStep, fusedStep);
+      chainOperations(reducerStartStep, fusedStep, Sets.<Graphs.Step>newHashSet());
       conf.set(
           BeamReducer.BEAM_REDUCER_KV_CODER,
           Base64.encodeBase64String(SerializableUtils.serializeToByteArray(
@@ -152,7 +154,7 @@ public class JobPrototype {
     }
 
     // Setup DoFns in BeamMapper.
-    chainOperations(startStep, fusedStep);
+    chainOperations(startStep, fusedStep, Sets.<Graphs.Step>newHashSet());
 
     job.setMapOutputKeyClass(BytesWritable.class);
     job.setMapOutputValueClass(byte[].class);
@@ -177,7 +179,8 @@ public class JobPrototype {
     return job;
   }
 
-  private void chainOperations(Graphs.Step current, Graphs.FusedStep fusedStep) {
+  private void chainOperations(
+      Graphs.Step current, Graphs.FusedStep fusedStep, Set<Graphs.Step> visited) {
     Operation<?> operation = current.getOperation();
     List<Graphs.Tag> outputTags = fusedStep.getOutputTags(current);
     for (Graphs.Tag outTag : outputTags) {
@@ -185,9 +188,12 @@ public class JobPrototype {
         operation.attachConsumer(outTag.getTupleTag(), consumer.getOperation());
       }
     }
+    visited.add(current);
     for (Graphs.Tag outTag : outputTags) {
       for (Graphs.Step consumer : fusedStep.getConsumers(outTag)) {
-        chainOperations(consumer, fusedStep);
+        if (!visited.contains(consumer)) {
+          chainOperations(consumer, fusedStep, visited);
+        }
       }
     }
   }


Mime
View raw message