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/07 23:19:58 UTC
incubator-tinkerpop git commit: I created Spark which is a class that
maintains a HashMap of persisted RDDs to prevent Spark for garbage collecting
them. This makes it so we can treat Spark like a file system where each RDD
is saved indefinately.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1025 e45c29342 -> f26e472ad
I created Spark which is a class that maintains a HashMap of persisted RDDs to prevent Spark for garbage collecting them. This makes it so we can treat Spark like a file system where each RDD is saved indefinately.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/f26e472a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/f26e472a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/f26e472a
Branch: refs/heads/TINKERPOP-1025
Commit: f26e472adf84163cefb932f9fb001563cd7f69b1
Parents: e45c293
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Mon Dec 7 15:19:49 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Mon Dec 7 15:19:49 2015 -0700
----------------------------------------------------------------------
.../process/computer/SparkContextHelper.java | 6 +-
.../process/computer/SparkGraphComputer.java | 7 +-
.../gremlin/spark/structure/Spark.java | 101 +++++++++++++++++++
.../spark/structure/io/PersistedInputRDD.java | 17 +---
.../spark/structure/io/PersistedOutputRDD.java | 3 +
.../process/computer/LocalPropertyTest.java | 3 +-
.../gremlin/spark/structure/SparkTest.java | 101 +++++++++++++++++++
.../io/PersistedInputOutputRDDTest.java | 25 +++--
8 files changed, 236 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkContextHelper.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkContextHelper.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkContextHelper.java
index ed553c5..6ceaf9f 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkContextHelper.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkContextHelper.java
@@ -22,6 +22,7 @@ package org.apache.tinkerpop.gremlin.spark.process.computer;
import org.apache.commons.configuration.Configuration;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -33,7 +34,10 @@ public final class SparkContextHelper {
}
public static void tryToCloseContext(final JavaSparkContext context, final Configuration configuration) {
- if (context != null && !configuration.getBoolean(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, false))
+ if (context != null && !configuration.getBoolean(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, false)) {
context.close();
+ if (null != Spark.getContext())
+ Spark.close();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/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 73f9435..21a379a 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
@@ -46,12 +46,12 @@ import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.spark.process.computer.payload.ViewIncomingPayload;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.InputFormatRDD;
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;
@@ -144,6 +144,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
JavaSparkContext sparkContext = null;
try {
sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
+ Spark.create(sparkContext.sc()); // this is the context RDD holder that prevents GC
updateLocalConfiguration(sparkContext, sparkConfiguration);
// add the project jars to the cluster
this.loadJars(sparkContext, hadoopConfiguration);
@@ -191,7 +192,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
+ Spark.removeRDD(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);
@@ -229,7 +230,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)); // delete persisted output rdd if it exists
+ Spark.removeRDD(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/f26e472a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/Spark.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/Spark.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/Spark.java
new file mode 100644
index 0000000..5a13593
--- /dev/null
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/Spark.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.tinkerpop.gremlin.spark.structure;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+import org.apache.spark.rdd.RDD;
+import scala.collection.JavaConversions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class Spark {
+
+ private static SparkContext CONTEXT;
+ private static final Map<String, RDD<?>> NAME_TO_RDD = new ConcurrentHashMap<>();
+
+ private Spark() {
+ }
+
+ private Spark(final SparkContext sparkContext) {
+ CONTEXT = sparkContext;
+ }
+
+ public static SparkContext getContext() {
+ return CONTEXT;
+ }
+
+ public static void create(final Configuration configuration) {
+ final SparkConf sparkConf = new SparkConf();
+ configuration.getKeys().forEachRemaining(key -> sparkConf.set(key, configuration.getProperty(key).toString()));
+ sparkConf.setAppName("Spark RDD Utility");
+ CONTEXT = SparkContext.getOrCreate(sparkConf);
+ }
+
+ public static void create(final SparkContext sparkContext) {
+ CONTEXT = sparkContext;
+ }
+
+ public static void refresh() {
+ for (final RDD<?> rdd : JavaConversions.asJavaIterable(CONTEXT.persistentRdds().values())) {
+ if (null != rdd.name()) {
+ NAME_TO_RDD.put(rdd.name(), rdd);
+ }
+ }
+ }
+
+ public static RDD<?> getRDD(final String name) {
+ Spark.refresh();
+ if (!NAME_TO_RDD.containsKey(name))
+ throw new IllegalArgumentException("The named RDD does not exist in the Spark context: " + name);
+ return NAME_TO_RDD.get(name);
+ }
+
+ public static boolean hasRDD(final String name) {
+ Spark.refresh();
+ return NAME_TO_RDD.containsKey(name);
+ }
+
+ public static List<RDD<?>> getRDDs() {
+ Spark.refresh();
+ return new ArrayList<>(NAME_TO_RDD.values());
+ }
+
+ public static void removeRDD(final String name) {
+ if (null == name)
+ return;
+ Spark.refresh();
+ final RDD<?> rdd = NAME_TO_RDD.remove(name);
+ if (null != rdd)
+ rdd.unpersist(true);
+ }
+
+ public static void close() {
+ CONTEXT.stop();
+ NAME_TO_RDD.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/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 2d049fb..a59d556 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
@@ -23,15 +23,9 @@ import org.apache.commons.configuration.Configuration;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.rdd.RDD;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
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;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -40,15 +34,14 @@ public final class PersistedInputRDD implements InputRDD {
@Override
public JavaPairRDD<Object, VertexWritable> readGraphRDD(final Configuration configuration, final JavaSparkContext sparkContext) {
+ Spark.create(sparkContext.sc());
final String inputRDDName = configuration.getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION, null);
if (null == inputRDDName)
throw new IllegalArgumentException(PersistedInputRDD.class.getSimpleName() + " requires " + Constants.GREMLIN_HADOOP_INPUT_LOCATION + " in order to retrieve the named graphRDD from the SparkContext");
- if (!PersistedInputRDD.getPersistedRDD(sparkContext, inputRDDName).isPresent())
- throw new IllegalArgumentException("The provided graphRDD name is not in the persisted RDDs of the SparkContext: " + inputRDDName);
- return JavaPairRDD.fromJavaRDD((JavaRDD) PersistedInputRDD.getPersistedRDD(sparkContext, inputRDDName).get().toJavaRDD());
+ return JavaPairRDD.fromJavaRDD((JavaRDD) Spark.getRDD(inputRDDName).toJavaRDD());
}
- public static Optional<RDD<?>> getPersistedRDD(final JavaSparkContext sparkContext, final String rddName) {
+ /*public static Optional<RDD<?>> getPersistedRDD(final JavaSparkContext sparkContext, final String rddName) {
final Iterator<Tuple2<Object, RDD<?>>> iterator = JavaSparkContext.toSparkContext(sparkContext).
getPersistentRDDs().
toList().iterator();
@@ -75,5 +68,5 @@ public final class PersistedInputRDD implements InputRDD {
for (final Object rddId : matchingRDDs) {
JavaSparkContext.toSparkContext(sparkContext).persistentRdds().remove(rddId).get().unpersist(false);
}
- }
+ }*/
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/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 146bc73..93451b8 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
@@ -23,6 +23,7 @@ import org.apache.commons.configuration.Configuration;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,5 +47,7 @@ public final class PersistedOutputRDD implements OutputRDD {
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");
+
+ Spark.refresh();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/LocalPropertyTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/LocalPropertyTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/LocalPropertyTest.java
index b27b7b9..9aa52d9 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/LocalPropertyTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/LocalPropertyTest.java
@@ -34,6 +34,7 @@ import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexPr
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.AbstractSparkTest;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedInputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
@@ -77,7 +78,7 @@ public class LocalPropertyTest extends AbstractSparkTest {
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
JavaSparkStatusTracker statusTracker = sparkContext.statusTracker();
assertTrue(statusTracker.getJobIdsForGroup("22").length >= 1);
- assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ assertTrue(Spark.hasRDD(rddName));
///////
configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, PersistedInputRDD.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/SparkTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/SparkTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/SparkTest.java
new file mode 100644
index 0000000..ea1fb2e
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/SparkTest.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.tinkerpop.gremlin.spark.structure;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.spark.rdd.RDD;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
+import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
+import org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD;
+import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.junit.Test;
+import scala.collection.JavaConversions;
+
+import static org.junit.Assert.*;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class SparkTest extends AbstractSparkTest {
+
+ @Test
+ public void testSparkRDDPersistence() throws Exception {
+ final String prefix = "target/test-output/graphRDD-";
+ final Configuration configuration = new BaseConfiguration();
+ configuration.setProperty("spark.master", "local[4]");
+ Spark.create(configuration);
+
+ configuration.setProperty("spark.serializer", GryoSerializer.class.getCanonicalName());
+ configuration.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, SparkHadoopGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
+ configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, GryoInputFormat.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, PersistedOutputRDD.class.getCanonicalName());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
+ configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
+
+ for (int i = 0; i < 10; i++) {
+ assertEquals(i, Spark.getRDDs().size());
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, prefix + i);
+ Graph graph = GraphFactory.open(configuration);
+ graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().iterations(1).create(graph)).submit().get();
+ assertNotNull(Spark.getRDD(prefix + i));
+ assertEquals(i + 1, Spark.getRDDs().size());
+ }
+
+ for (int i = 9; i >= 0; i--) {
+ assertEquals(i + 1, getPersistedRDDSize());
+ assertEquals(i + 1, Spark.getRDDs().size());
+ assertTrue(hasPersistedRDD(prefix + i));
+ Spark.removeRDD(prefix + i);
+ assertFalse(hasPersistedRDD(prefix + i));
+ }
+
+ assertEquals(0, getPersistedRDDSize());
+ assertEquals(0, Spark.getRDDs().size());
+ Spark.close();
+ }
+
+ private static boolean hasPersistedRDD(final String name) {
+ for (final RDD<?> rdd : JavaConversions.asJavaIterable(Spark.getContext().persistentRdds().values())) {
+ if (null != rdd.name() && rdd.name().equals(name))
+ return true;
+ }
+ return false;
+ }
+
+ private static int getPersistedRDDSize() {
+ int counter = 0;
+ for (final RDD<?> rdd : JavaConversions.asJavaIterable(Spark.getContext().persistentRdds().values())) {
+ if (null != rdd.name())
+ counter++;
+ }
+ return counter;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f26e472a/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 d99e287..13aa2a9 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
@@ -37,6 +37,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.engine.ComputerTraversalEn
import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.io.IoCore;
@@ -44,7 +45,6 @@ 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.*;
@@ -80,7 +80,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
sparkConfiguration.setAppName("shouldNotPersistRDDAcrossJobs");
ConfUtil.makeHadoopConfiguration(configuration).forEach(entry -> sparkConfiguration.set(entry.getKey(), entry.getValue()));
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertFalse(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ Spark.create(sparkContext.sc());
+ assertFalse(Spark.hasRDD(rddName));
}
@Test
@@ -109,7 +110,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
sparkConfiguration.setAppName("shouldPersistRDDAcrossJobs");
ConfUtil.makeHadoopConfiguration(configuration).forEach(entry -> sparkConfiguration.set(entry.getKey(), entry.getValue()));
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ Spark.create(sparkContext.sc());
+ assertTrue(Spark.hasRDD(rddName));
///////
configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, PersistedInputRDD.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName);
@@ -158,7 +160,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
SparkConf sparkConfiguration = new SparkConf();
sparkConfiguration.setAppName("testBulkLoaderVertexProgramChain");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ Spark.create(sparkContext.sc());
+ assertTrue(Spark.hasRDD(rddName));
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
@@ -199,7 +202,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
SparkConf sparkConfiguration = new SparkConf();
sparkConfiguration.setAppName("testBulkLoaderVertexProgramChain");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ Spark.create(sparkContext.sc());
+ assertTrue(Spark.hasRDD(rddName));
////
final Graph graph = TinkerGraph.open();
final GraphTraversalSource g = graph.traversal();
@@ -235,7 +239,8 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
SparkConf sparkConfiguration = new SparkConf();
sparkConfiguration.setAppName("testComplexChain");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
- assertTrue(PersistedInputRDD.getPersistedRDD(sparkContext, rddName).isPresent());
+ Spark.create(sparkContext.sc());
+ assertTrue(Spark.hasRDD(rddName));
////
configuration.setProperty(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, PersistedInputRDD.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName);
@@ -250,7 +255,7 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
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());
+ assertTrue(Spark.hasRDD(rddName));
////
graph = GraphFactory.open(configuration);
graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
@@ -260,17 +265,17 @@ public class PersistedInputOutputRDDTest extends AbstractSparkTest {
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());
+ assertTrue(Spark.hasRDD(rddName));
////
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());
+ assertFalse(Spark.hasRDD(rddName));
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());
+ assertFalse(Spark.hasRDD(rddName));
}
}