You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2015/03/31 00:00:52 UTC
incubator-tinkerpop git commit: in case there are messages sent to
non-existent vertices. filter the Spark payload.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master d8d5cbed5 -> e02c8cc96
in case there are messages sent to non-existent vertices. filter the Spark payload.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e02c8cc9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e02c8cc9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e02c8cc9
Branch: refs/heads/master
Commit: e02c8cc963431665a71ac05f746b99e7d6666c26
Parents: d8d5cbe
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Mon Mar 30 16:00:08 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Mon Mar 30 16:00:37 2015 -0600
----------------------------------------------------------------------
.../process/computer/spark/SparkExecutor.java | 33 +++++++++++---------
.../computer/spark/SparkGraphComputer.java | 4 +--
2 files changed, 20 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e02c8cc9/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
index ec1dd6f..7e6eed3 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
@@ -84,19 +84,22 @@ public final class SparkExecutor {
// "message pass" by merging the message payloads with the vertex payloads
final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration).getMessageCombiner().orElse(null);
- final JavaPairRDD<Object, SparkVertexPayload<M>> verticesHoldingIncomingMessages = (JavaPairRDD) verticesAndOutgoingMessages.reduceByKey((payloadA, payloadB) -> {
- if (payloadA.isVertex()) {
- if (payloadB.isVertex())
- throw new IllegalStateException("It should not be the case that two vertices reduce to the same key: " + payloadA.asVertexPayload().getVertex() + "==" + payloadB.asVertexPayload().getVertex());
- else {
- payloadA.addMessages(payloadB.getMessages(), messageCombiner);
- return payloadA;
- }
- } else {
- payloadB.addMessages(payloadA.getMessages(), messageCombiner);
- return payloadB;
- }
- });
+ final JavaPairRDD<Object, SparkVertexPayload<M>> verticesHoldingIncomingMessages = (JavaPairRDD) verticesAndOutgoingMessages
+ .reduceByKey((payloadA, payloadB) -> {
+ if (payloadA.isVertex()) {
+ if (payloadB.isVertex())
+ throw new IllegalStateException("It should not be the case that two vertices reduce to the same key: " + payloadA.asVertexPayload().getVertex() + "==" + payloadB.asVertexPayload().getVertex());
+ else {
+ payloadA.addMessages(payloadB.getMessages(), messageCombiner);
+ return payloadA;
+ }
+ } else {
+ payloadB.addMessages(payloadA.getMessages(), messageCombiner);
+ return payloadB;
+ }
+ })
+ .filter(payload -> payload._2().isVertex()); // just in case there are messages sent to vertices that do not exist
+
verticesHoldingIncomingMessages.foreachPartition(partitionIterator -> {
}); // need to complete a task so its BSP.
return verticesHoldingIncomingMessages;
@@ -107,8 +110,8 @@ public final class SparkExecutor {
/////////////////////////////
/////////////////////////////
- public static <M> JavaPairRDD<Object, Tuple2<List<DetachedVertexProperty<Object>>, List<M>>> executeVertexProgramIteration2(final JavaPairRDD<Object, VertexWritable> graphRDD, final JavaPairRDD<Object, Tuple2<List<DetachedVertexProperty<Object>>,List<M>>> viewMessageRDD, final SparkMemory memory, final Configuration apacheConfiguration) {
- final JavaPairRDD<Object, Tuple2<VertexWritable, Optional<Tuple2<List<DetachedVertexProperty<Object>>,List<M>>>>> graphViewMessagesRDD = graphRDD.leftOuterJoin(viewMessageRDD);
+ public static <M> JavaPairRDD<Object, Tuple2<List<DetachedVertexProperty<Object>>, List<M>>> executeVertexProgramIteration2(final JavaPairRDD<Object, VertexWritable> graphRDD, final JavaPairRDD<Object, Tuple2<List<DetachedVertexProperty<Object>>, List<M>>> viewMessageRDD, final SparkMemory memory, final Configuration apacheConfiguration) {
+ final JavaPairRDD<Object, Tuple2<VertexWritable, Optional<Tuple2<List<DetachedVertexProperty<Object>>, List<M>>>>> graphViewMessagesRDD = graphRDD.leftOuterJoin(viewMessageRDD);
final JavaPairRDD<Object, Tuple2<List<DetachedVertexProperty<Object>>, List<Tuple2<Object, M>>>> viewAndMessagesRDD = graphViewMessagesRDD.mapPartitionsToPair(partitions -> {
final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e02c8cc9/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
index eb61d12..e5388c9 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
@@ -164,7 +164,7 @@ public final class SparkGraphComputer implements GraphComputer {
NullWritable.class,
VertexWritable.class)
.mapToPair(tuple -> new Tuple2<>(tuple._2().get().id(), new SparkVertexPayload<>(tuple._2().get())))
- .reduceByKey((a, b) -> a).cache(); // partition the graph across the cluster // todo: cache?
+ .reduceByKey((a, b) -> a); // partition the graph across the cluster
////////////////////////////////
// process the vertex program //
@@ -182,7 +182,7 @@ public final class SparkGraphComputer implements GraphComputer {
// execute the vertex program
while (true) {
memory.setInTask(true);
- graphRDD = SparkExecutor.executeVertexProgramIteration(graphRDD, memory, vertexProgramConfiguration).cache();
+ graphRDD = SparkExecutor.executeVertexProgramIteration(graphRDD, memory, vertexProgramConfiguration);
memory.setInTask(false);
if (this.vertexProgram.terminate(memory))
break;