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/12/04 16:36:24 UTC
incubator-tinkerpop git commit: mapReduceRDD makes use of a post view
merge. @dalaro realized this was important prior to graph writing. Thus,
moved the view merge to pre-mapreduce and pre-graph output. Added more
rigorous property checking to PageRankVe
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1025 [created] 5c7bc38bd
mapReduceRDD makes use of a post view merge. @dalaro realized this was important prior to graph writing. Thus, moved the view merge to pre-mapreduce and pre-graph output. Added more rigorous property checking to PageRankVertexProgramTest. InputFormatRDD and ToyGraphInputRDD no longer require reduceByKey() initiation because of merged veiws.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/5c7bc38b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/5c7bc38b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/5c7bc38b
Branch: refs/heads/TINKERPOP-1025
Commit: 5c7bc38bdb42ae50243f58a22fc74bc094be6333
Parents: 04f7e24
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri Dec 4 08:36:08 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri Dec 4 08:36:08 2015 -0700
----------------------------------------------------------------------
.../ranking/PageRankVertexProgramTest.java | 5 ++++
.../spark/process/computer/SparkExecutor.java | 29 ++++++++------------
.../process/computer/SparkGraphComputer.java | 17 ++++++++----
.../spark/structure/io/InputFormatRDD.java | 3 +-
.../spark/structure/io/PersistedInputRDD.java | 19 +++++++++++++
.../spark/structure/io/PersistedOutputRDD.java | 1 +
.../computer/SparkHadoopGraphProvider.java | 2 --
.../io/PersistedInputOutputRDDTest.java | 4 +--
.../spark/structure/io/ToyGraphInputRDD.java | 2 +-
9 files changed, 53 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
index df65c7b..c86f875 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
@@ -23,6 +23,7 @@ import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.junit.Test;
import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
@@ -40,8 +41,12 @@ public class PageRankVertexProgramTest extends AbstractGremlinProcessTest {
if (g.getGraphComputer().get().features().supportsResultGraphPersistCombination(GraphComputer.ResultGraph.NEW, GraphComputer.Persist.VERTEX_PROPERTIES)) {
final ComputerResult result = graph.compute(g.getGraphComputer().get().getClass()).program(PageRankVertexProgram.build().create(graph)).submit().get();
result.graph().traversal().V().forEachRemaining(v -> {
+ assertEquals(4, v.keys().size()); // name, age/lang, edgeCount, pageRank
assertTrue(v.keys().contains("name"));
+ assertTrue(v.keys().contains(PageRankVertexProgram.EDGE_COUNT));
assertTrue(v.keys().contains(PageRankVertexProgram.PAGE_RANK));
+ assertEquals(1, IteratorUtils.count(v.values("name")));
+ assertEquals(1, IteratorUtils.count(v.values(PageRankVertexProgram.PAGE_RANK)));
final String name = v.value("name");
final Double pageRank = v.value(PageRankVertexProgram.PAGE_RANK);
//System.out.println(name + "-----" + pageRank);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
index 0350101..0a45b37 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java
@@ -138,27 +138,22 @@ public final class SparkExecutor {
return newViewIncomingRDD;
}
+ public static <M> JavaPairRDD<Object, VertexWritable> prepareFinalGraphRDD(final JavaPairRDD<Object, VertexWritable> graphRDD, final JavaPairRDD<Object, ViewIncomingPayload<M>> viewIncomingRDD, final String[] elementComputeKeys) {
+ // attach the final computed view to the cached graph
+ return graphRDD.leftOuterJoin(viewIncomingRDD)
+ .mapValues(tuple -> {
+ final StarGraph.StarVertex vertex = tuple._1().get();
+ vertex.dropVertexProperties(elementComputeKeys);
+ final List<DetachedVertexProperty<Object>> view = tuple._2().isPresent() ? tuple._2().get().getView() : Collections.emptyList();
+ view.forEach(property -> property.attach(Attachable.Method.create(vertex)));
+ return tuple._1();
+ });
+ }
+
/////////////////
// MAP REDUCE //
////////////////
- public static <M> JavaPairRDD<Object, VertexWritable> prepareGraphRDDForMapReduce(final JavaPairRDD<Object, VertexWritable> graphRDD, final JavaPairRDD<Object, ViewIncomingPayload<M>> viewIncomingRDD, final String[] elementComputeKeys) {
- return (null == viewIncomingRDD) ? // there was no vertex program
- graphRDD.mapValues(vertexWritable -> {
- vertexWritable.get().dropEdges();
- return vertexWritable;
- }) :
- graphRDD.leftOuterJoin(viewIncomingRDD)
- .mapValues(tuple -> {
- final StarGraph.StarVertex vertex = tuple._1().get();
- vertex.dropEdges();
- vertex.dropVertexProperties(elementComputeKeys);
- final List<DetachedVertexProperty<Object>> view = tuple._2().isPresent() ? tuple._2().get().getView() : Collections.emptyList();
- view.forEach(property -> property.attach(Attachable.Method.create(vertex)));
- return tuple._1();
- });
- }
-
public static <K, V> JavaPairRDD<K, V> executeMap(final JavaPairRDD<Object, VertexWritable> graphRDD, final MapReduce<K, V, ?, ?, ?> mapReduce, final Configuration apacheConfiguration) {
JavaPairRDD<K, V> mapRDD = graphRDD.mapPartitionsToPair(partitionIterator -> {
HadoopPools.initialize(apacheConfiguration);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
index c20f1b0..715eb36 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
@@ -51,6 +51,8 @@ import org.apache.tinkerpop.gremlin.spark.structure.io.InputOutputHelper;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.OutputFormatRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.OutputRDD;
+import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedInputRDD;
+import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
import java.io.File;
import java.io.IOException;
@@ -146,12 +148,11 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// add the project jars to the cluster
this.loadJars(sparkContext, hadoopConfiguration);
// create a message-passing friendly rdd from the input rdd
- final JavaPairRDD<Object, VertexWritable> graphRDD;
+ JavaPairRDD<Object, VertexWritable> graphRDD;
try {
graphRDD = hadoopConfiguration.getClass(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, InputFormatRDD.class, InputRDD.class)
.newInstance()
.readGraphRDD(apacheConfiguration, sparkContext)
- .setName(sparkConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "graphRDD"))
.cache();
} catch (final InstantiationException | IllegalAccessException e) {
throw new IllegalStateException(e.getMessage(), e);
@@ -184,6 +185,8 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
}
}
// write the graph rdd using the output rdd
+ final String[] elementComputeKeys = this.vertexProgram == null ? new String[0] : this.vertexProgram.getElementComputeKeys().toArray(new String[this.vertexProgram.getElementComputeKeys().size()]);
+ graphRDD = SparkExecutor.prepareFinalGraphRDD(graphRDD, viewIncomingRDD, elementComputeKeys);
if ((hadoopConfiguration.get(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, null) != null ||
hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, null) != null) &&
!this.persist.equals(GraphComputer.Persist.NOTHING)) {
@@ -203,8 +206,10 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// process the map reducers //
//////////////////////////////
if (!this.mapReducers.isEmpty()) {
- final String[] elementComputeKeys = this.vertexProgram == null ? new String[0] : this.vertexProgram.getElementComputeKeys().toArray(new String[this.vertexProgram.getElementComputeKeys().size()]);
- final JavaPairRDD<Object, VertexWritable> mapReduceGraphRDD = SparkExecutor.prepareGraphRDDForMapReduce(graphRDD, viewIncomingRDD, elementComputeKeys).setName("mapReduceGraphRDD").cache();
+ final JavaPairRDD<Object, VertexWritable> mapReduceGraphRDD = graphRDD.mapValues(vertexWritable -> {
+ vertexWritable.get().dropEdges();
+ return vertexWritable;
+ }).setName("mapReduceGraphRDD").cache(); // drop all the edges of the graph as they are not used in mapReduce processing
for (final MapReduce mapReduce : this.mapReducers) {
// execute the map reduce job
final HadoopConfiguration newApacheConfiguration = new HadoopConfiguration(apacheConfiguration);
@@ -221,8 +226,10 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
}
// unpersist the graphRDD if it will no longer be used
- if (hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, null) == null || this.persist.equals(GraphComputer.Persist.NOTHING)) {
+ if (!PersistedOutputRDD.class.equals(hadoopConfiguration.getClass(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, null)) || this.persist.equals(GraphComputer.Persist.NOTHING)) {
graphRDD.unpersist();
+ PersistedInputRDD.removePersistedRDD(sparkContext,hadoopConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION));
+ // PersistedInputRDD.removePersistedRDD(sparkContext,hadoopConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION)); // DO WE WANT CHAINS TO DIE OR SAVE LIKE A FILE SYSTEM?
}
// update runtime and return the newly computed graph
finalMemory.setRuntime(System.currentTimeMillis() - startTime);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputFormatRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputFormatRDD.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputFormatRDD.java
index f768939..72fe57b 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputFormatRDD.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputFormatRDD.java
@@ -41,7 +41,6 @@ public final class InputFormatRDD implements InputRDD {
(Class<InputFormat<NullWritable, VertexWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class),
NullWritable.class,
VertexWritable.class)
- .mapToPair(tuple -> new Tuple2<>(tuple._2().get().id(), new VertexWritable(tuple._2().get())))
- .reduceByKey((a, b) -> a); // if this is not done, then the graph is partitioned and you can have duplicate vertices
+ .mapToPair(tuple -> new Tuple2<>(tuple._2().get().id(), new VertexWritable(tuple._2().get())));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputRDD.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputRDD.java
index e3c27ec..0130968 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputRDD.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputRDD.java
@@ -29,6 +29,8 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import scala.Tuple2;
import scala.collection.Iterator;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
/**
@@ -57,4 +59,21 @@ public final class PersistedInputRDD implements InputRDD {
}
return Optional.empty();
}
+
+ public static void removePersistedRDD(final JavaSparkContext sparkContext, final String rddName) {
+ if (null == rddName)
+ return;
+ final List<Object> matchingRDDs = new ArrayList<>();
+ final Iterator<Tuple2<Object, RDD<?>>> iterator = JavaSparkContext.toSparkContext(sparkContext).
+ getPersistentRDDs().
+ toList().iterator();
+ while (iterator.hasNext()) {
+ final Tuple2<Object, RDD<?>> tuple2 = iterator.next();
+ if (tuple2._2().toString().contains(rddName))
+ matchingRDDs.add(tuple2._1());
+ }
+ for (final Object rddId : matchingRDDs) {
+ JavaSparkContext.toSparkContext(sparkContext).persistentRdds().remove(rddId);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
index abeeaaa..bb43411 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedOutputRDD.java
@@ -35,6 +35,7 @@ public final class PersistedOutputRDD implements OutputRDD {
@Override
public void writeGraphRDD(final Configuration configuration, final JavaPairRDD<Object, VertexWritable> graphRDD) {
+ graphRDD.setName(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "graphRDD")).cache();
if (!configuration.getBoolean(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, false))
LOGGER.warn("The SparkContext should be persisted in order for the RDD to persist across jobs. To do so, set " + Constants.GREMLIN_SPARK_PERSIST_CONTEXT + " to true");
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
index 618904a..bf4f63e 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
@@ -23,10 +23,8 @@ import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.HadoopGraphProvider;
import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgramTest;
-import org.apache.tinkerpop.gremlin.process.computer.ranking.PageRankVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.engine.ComputerTraversalEngine;
-import org.apache.tinkerpop.gremlin.spark.structure.io.InputRDDFormat;
import org.apache.tinkerpop.gremlin.spark.structure.io.ToyGraphInputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
import org.apache.tinkerpop.gremlin.structure.Graph;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDTest.java
index bc55c04..2e91609 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDTest.java
@@ -157,7 +157,7 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
SparkConf sparkConfiguration = new SparkConf();
sparkConfiguration.setAppName("testBulkLoaderVertexProgramChain");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertFalse(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
@@ -200,7 +200,7 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
SparkConf sparkConfiguration = new SparkConf();
sparkConfiguration.setAppName("testBulkLoaderVertexProgramChain");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertFalse(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5c7bc38b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/ToyGraphInputRDD.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/ToyGraphInputRDD.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/ToyGraphInputRDD.java
index ff253e3..ea3636f 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/ToyGraphInputRDD.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/ToyGraphInputRDD.java
@@ -68,6 +68,6 @@ public final class ToyGraphInputRDD implements InputRDD {
} else
throw new IllegalArgumentException("No legal toy graph was provided to load: " + configuration.getProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
- return sparkContext.parallelize(vertices).mapToPair(vertex -> new Tuple2<>(vertex.get().id(), vertex)).reduceByKey((a, b) -> a); // if this is not done, then the graph is partitioned and you can have duplicate vertices;
+ return sparkContext.parallelize(vertices).mapToPair(vertex -> new Tuple2<>(vertex.get().id(), vertex));
}
}