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:17 UTC
[06/13] incubator-tinkerpop git commit: better control over RDD
persistence.
better control over RDD persistence.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/d17224c4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/d17224c4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/d17224c4
Branch: refs/heads/master
Commit: d17224c4a310ed89d15ff04142d36b078d00e677
Parents: 63446c2
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Mon Dec 7 15:48:00 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Mon Dec 7 15:48:00 2015 -0700
----------------------------------------------------------------------
.../process/computer/SparkContextHelper.java | 5 ++--
.../process/computer/SparkGraphComputer.java | 3 +-
.../spark/structure/io/PersistedInputRDD.java | 31 +-------------------
.../spark/structure/io/PersistedOutputRDD.java | 1 +
.../gremlin/spark/AbstractSparkTest.java | 3 ++
5 files changed, 8 insertions(+), 35 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d17224c4/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 6ceaf9f..488186a 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
@@ -35,9 +35,8 @@ 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)) {
- context.close();
- if (null != Spark.getContext())
- Spark.close();
+ Spark.create(context.sc());
+ Spark.close();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d17224c4/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 21a379a..0c8ac74 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
@@ -192,7 +192,6 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
hadoopConfiguration.get(Constants.GREMLIN_SPARK_GRAPH_OUTPUT_RDD, null) != null) &&
!this.persist.equals(GraphComputer.Persist.NOTHING)) {
try {
- 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);
@@ -211,7 +210,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
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
+ }).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);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d17224c4/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 a59d556..88e1efa 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
@@ -34,39 +34,10 @@ 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");
+ Spark.create(sparkContext.sc());
return JavaPairRDD.fromJavaRDD((JavaRDD) Spark.getRDD(inputRDDName).toJavaRDD());
}
-
- /*public static Optional<RDD<?>> getPersistedRDD(final JavaSparkContext sparkContext, final String rddName) {
- 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))
- return Optional.of(tuple2._2());
- }
- 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).get().unpersist(false);
- }
- }*/
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d17224c4/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 93451b8..5188f9a 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
@@ -38,6 +38,7 @@ public final class PersistedOutputRDD implements OutputRDD {
public void writeGraphRDD(final Configuration configuration, final JavaPairRDD<Object, VertexWritable> graphRDD) {
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");
+ Spark.removeRDD(configuration.getString(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)); // this might be bad cause it unpersists the job RDD
if (!configuration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true))
graphRDD.mapValues(vertex -> {
vertex.get().dropEdges();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/d17224c4/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/AbstractSparkTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/AbstractSparkTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/AbstractSparkTest.java
index 02b0e18..ccff1ab 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/AbstractSparkTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/AbstractSparkTest.java
@@ -22,6 +22,7 @@ package org.apache.tinkerpop.gremlin.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.junit.After;
import org.junit.Before;
@@ -38,6 +39,8 @@ public abstract class AbstractSparkTest {
sparkConfiguration.set("spark.master", "local[4]");
JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
sparkContext.close();
+ Spark.create(sparkContext.sc());
+ Spark.close();
System.out.println("SparkContext has been closed for " + this.getClass().getCanonicalName() + "-setupTest");
}
}