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/08 14:47:13 UTC
[02/13] incubator-tinkerpop git commit: numerous tweaks trying to get
things clean and clear. Added more tests to PersistedInputOutputRDDTest that
show good long chain vertex programs with various degrees of Persist and
Hadoop OLTP access, etc. Looking g
numerous tweaks trying to get things clean and clear. Added more tests to PersistedInputOutputRDDTest that show good long chain vertex programs with various degrees of Persist and Hadoop OLTP access, etc. Looking good. Still BulkLoaderVertexProgram problem with InputRDD... don't know what the problem is still (unfortunately).
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/e45c2934
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/e45c2934
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/e45c2934
Branch: refs/heads/master
Commit: e45c293425ed4d9c317b5efbb3a81a9874f7e0e6
Parents: 5c7bc38
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri Dec 4 11:14:50 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri Dec 4 11:14:50 2015 -0700
----------------------------------------------------------------------
.../BulkLoaderVertexProgramTest.java | 10 +--
.../process/computer/SparkGraphComputer.java | 4 +-
.../spark/structure/io/InputOutputHelper.java | 2 +
.../spark/structure/io/PersistedInputRDD.java | 2 +-
.../spark/structure/io/PersistedOutputRDD.java | 10 ++-
.../computer/SparkHadoopGraphProvider.java | 1 +
.../io/PersistedInputOutputRDDTest.java | 80 +++++++++++++++++---
7 files changed, 90 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
index 992bfa4..ccfd699 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/bulkloading/BulkLoaderVertexProgramTest.java
@@ -94,7 +94,7 @@ public class BulkLoaderVertexProgramTest extends AbstractGremlinProcessTest {
.writeGraph(getWriteGraphConfiguration()).create(graph);
final BulkLoader loader = getBulkLoader(blvp);
assertFalse(loader.useUserSuppliedIds());
- graph.compute(g.getGraphComputer().get().getClass()).workers(1).program(blvp).submit().get();
+ graph.compute(graphComputerClass.get()).workers(1).program(blvp).submit().get();
assertGraphEquality(graph, getWriteGraph(), v -> v.value(loader.getVertexIdProperty()));
}
@@ -106,7 +106,7 @@ public class BulkLoaderVertexProgramTest extends AbstractGremlinProcessTest {
.writeGraph(getWriteGraphConfiguration()).create(graph);
final BulkLoader loader = getBulkLoader(blvp);
assertTrue(loader.useUserSuppliedIds());
- graph.compute(g.getGraphComputer().get().getClass()).workers(1).program(blvp).submit().get();
+ graph.compute(graphComputerClass.get()).workers(1).program(blvp).submit().get();
assertGraphEquality(graph, getWriteGraph());
}
@@ -116,8 +116,8 @@ public class BulkLoaderVertexProgramTest extends AbstractGremlinProcessTest {
final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
.userSuppliedIds(true)
.writeGraph(getWriteGraphConfiguration()).create(graph);
- graph.compute(g.getGraphComputer().get().getClass()).workers(1).program(blvp).submit().get(); // initial
- graph.compute(g.getGraphComputer().get().getClass()).workers(1).program(blvp).submit().get(); // incremental
+ graph.compute(graphComputerClass.get()).workers(1).program(blvp).submit().get(); // initial
+ graph.compute(graphComputerClass.get()).workers(1).program(blvp).submit().get(); // incremental
assertGraphEquality(graph, getWriteGraph());
}
@@ -129,7 +129,7 @@ public class BulkLoaderVertexProgramTest extends AbstractGremlinProcessTest {
final BulkLoaderVertexProgram blvp = BulkLoaderVertexProgram.build()
.userSuppliedIds(true)
.writeGraph(getWriteGraphConfiguration()).create(graph);
- graph.compute(g.getGraphComputer().get().getClass()).workers(1).program(blvp).submit().get();
+ graph.compute(graphComputerClass.get()).workers(1).program(blvp).submit().get();
assertGraphEquality(graph, getWriteGraph());
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/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 715eb36..73f9435 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
@@ -191,6 +191,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, null) != null) &&
!this.persist.equals(GraphComputer.Persist.NOTHING)) {
try {
+ PersistedInputRDD.removePersistedRDD(sparkContext, hadoopConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)); // delete existing persisted output rdd
hadoopConfiguration.getClass(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, OutputFormatRDD.class, OutputRDD.class)
.newInstance()
.writeGraphRDD(apacheConfiguration, graphRDD);
@@ -228,8 +229,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
// unpersist the graphRDD if it will no longer be used
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?
+ PersistedInputRDD.removePersistedRDD(sparkContext, hadoopConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)); // delete persisted output rdd if it exists
}
// update runtime and return the newly computed graph
finalMemory.setRuntime(System.currentTimeMillis() - startTime);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputOutputHelper.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputOutputHelper.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputOutputHelper.java
index 1e279de..6736fb0 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputOutputHelper.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputOutputHelper.java
@@ -68,6 +68,8 @@ public final class InputOutputHelper {
final BaseConfiguration newConfiguration = new BaseConfiguration();
newConfiguration.copy(org.apache.tinkerpop.gremlin.hadoop.structure.io.InputOutputHelper.getOutputGraph(configuration, resultGraph, persist).configuration());
if (resultGraph.equals(GraphComputer.ResultGraph.NEW) && hadoopConfiguration.containsKey(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD)) {
+ newConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputRDDFormat.class.getCanonicalName());
+ //newConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputRDDFormat.class.getCanonicalName());
newConfiguration.setProperty(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, InputOutputHelper.getInputFormat((Class) Class.forName(hadoopConfiguration.getString(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD))).getCanonicalName());
if (newConfiguration.getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION, "").endsWith("/" + Constants.HIDDEN_G)) { // Spark RDDs are not namespaced the same as Hadoop
newConfiguration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, newConfiguration.getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION).replace("/" + Constants.HIDDEN_G, ""));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/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 0130968..2d049fb 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
@@ -73,7 +73,7 @@ public final class PersistedInputRDD implements InputRDD {
matchingRDDs.add(tuple2._1());
}
for (final Object rddId : matchingRDDs) {
- JavaSparkContext.toSparkContext(sparkContext).persistentRdds().remove(rddId);
+ JavaSparkContext.toSparkContext(sparkContext).persistentRdds().remove(rddId).get().unpersist(false);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/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 bb43411..146bc73 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,7 +35,15 @@ 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.containsKey(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION))
+ throw new IllegalArgumentException("There is no provided " + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + " to write the persisted RDD to");
+ if (!configuration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true))
+ graphRDD.mapValues(vertex -> {
+ vertex.get().dropEdges();
+ return vertex;
+ }).setName(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)).cache();
+ else
+ graphRDD.setName(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)).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/e45c2934/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 bf4f63e..f0ba08f 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
@@ -47,6 +47,7 @@ public final class SparkHadoopGraphProvider extends HadoopGraphProvider {
if (null != loadGraphWith &&
!test.equals(BulkLoaderVertexProgramTest.class) &&
RANDOM.nextBoolean()) {
+ // config.put(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputRDDFormat.class.getCanonicalName());
config.put(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, ToyGraphInputRDD.class.getCanonicalName());
}
/// spark configuration
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/e45c2934/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 2e91609..d99e287 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
@@ -44,6 +44,7 @@ import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
import org.junit.Test;
+import java.io.File;
import java.util.UUID;
import static org.junit.Assert.*;
@@ -142,7 +143,7 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
final Configuration writeConfiguration = new BaseConfiguration();
writeConfiguration.setProperty(Graph.GRAPH, TinkerGraph.class.getCanonicalName());
writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_FORMAT, "gryo");
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, "target/test-output/tinkergraph.kryo");
+ writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, "target/test-output/testBulkLoaderVertexProgramChain.kryo");
final Graph bulkLoaderGraph = pageRankGraph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().create(pageRankGraph)).submit().get().graph();
bulkLoaderGraph.compute(SparkGraphComputer.class)
.persist(GraphComputer.Persist.NOTHING)
@@ -161,14 +162,12 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
- graph.io(IoCore.gryo()).readGraph("target/test-output/tinkergraph.kryo");
+ graph.io(IoCore.gryo()).readGraph("target/test-output/testBulkLoaderVertexProgramChain.kryo");
assertEquals(6l, g.V().count().next().longValue());
- assertEquals(6l, g.E().count().next().longValue());
+ assertEquals(0l, g.E().count().next().longValue());
assertEquals("marko", g.V().has("name", "marko").values("name").next());
assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
- assertEquals(0l, g.E().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
assertEquals(6l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
- assertEquals(0l, g.E().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
}
@Test
@@ -189,8 +188,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
final Configuration writeConfiguration = new BaseConfiguration();
writeConfiguration.setProperty(Graph.GRAPH, TinkerGraph.class.getCanonicalName());
writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_FORMAT, "gryo");
- writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, "target/test-output/tinkergraph.kryo");
- final Graph bulkLoaderGraph = pageRankGraph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().create(pageRankGraph)).submit().get().graph();
+ writeConfiguration.setProperty(TinkerGraph.GREMLIN_TINKERGRAPH_GRAPH_LOCATION, "target/test-output/testBulkLoaderVertexProgramChainWithInputOutputHelperMapping.kryo");
+ final Graph bulkLoaderGraph = pageRankGraph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).program(PageRankVertexProgram.build().create(pageRankGraph)).submit().get().graph();
bulkLoaderGraph.compute(SparkGraphComputer.class)
.persist(GraphComputer.Persist.NOTHING)
.workers(1)
@@ -204,13 +203,74 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
- graph.io(IoCore.gryo()).readGraph("target/test-output/tinkergraph.kryo");
+ graph.io(IoCore.gryo()).readGraph("target/test-output/testBulkLoaderVertexProgramChainWithInputOutputHelperMapping.kryo");
assertEquals(6l, g.V().count().next().longValue());
assertEquals(6l, g.E().count().next().longValue());
assertEquals("marko", g.V().has("name", "marko").values("name").next());
assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
- assertEquals(0l, g.E().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
assertEquals(6l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
- assertEquals(0l, g.E().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
+ }
+
+ @Test
+ public void testComplexChain() throws Exception {
+ final String rddName = "target/test-output/graphRDD";
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty("spark.master", "local[4]");
+ configuration.setProperty("spark.serializer", GryoSerializer.class.getCanonicalName());
+ configuration.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
+ configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
+ configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
+ Graph graph = GraphFactory.open(configuration);
+ graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
+ GraphTraversalSource g = graph.traversal();
+ assertEquals(6l, g.V().count().next().longValue());
+ assertEquals(6l, g.E().count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
+ ////
+ SparkConf sparkConfiguration = new SparkConf();
+ sparkConfiguration.setAppName("testComplexChain");
+ JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
+ assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ ////
+ configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, PersistedInputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName);
+ configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
+ ////
+ graph = GraphFactory.open(configuration);
+ graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
+ g = graph.traversal();
+ assertEquals(6l, g.V().count().next().longValue());
+ assertEquals(6l, g.E().count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
+ ////
+ assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ ////
+ graph = GraphFactory.open(configuration);
+ graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
+ g = graph.traversal();
+ assertEquals(6l, g.V().count().next().longValue());
+ assertEquals(0l, g.E().count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
+ assertEquals(6l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
+ ////
+ assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ ////
+ graph = GraphFactory.open(configuration);
+ graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.NOTHING).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
+ assertFalse(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ g = graph.traversal();
+ assertEquals(0l, g.V().count().next().longValue());
+ assertEquals(0l, g.E().count().next().longValue());
+ assertEquals(0l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
+ assertEquals(0l, g.V().values(PageRankVertexProgram.EDGE_COUNT).count().next().longValue());
+ ////
+ assertFalse(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
}
}