You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2015/12/15 23:58:43 UTC
[04/44] incubator-tinkerpop git commit: added Spark persited RDD
utility that can be spark.ls(), spark.head(), spark.rm(), spark.describe(),
etc. in the Console. Really cool. Added a SparkGremlinPluginTest that verifies
everything works as expected. Upda
added Spark persited RDD utility that can be spark.ls(), spark.head(), spark.rm(), spark.describe(), etc. in the Console. Really cool. Added a SparkGremlinPluginTest that verifies everything works as expected. Updated docs explaining the new tool.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/f8fabe20
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/f8fabe20
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/f8fabe20
Branch: refs/heads/TINKERPOP-1035
Commit: f8fabe20108f4cec8f4c50c7f7bf6523c112acac
Parents: 4d1d8c9
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Dec 8 09:28:19 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Dec 8 09:28:19 2015 -0700
----------------------------------------------------------------------
docs/src/reference/implementations.asciidoc | 31 ++++-
hadoop-gremlin/conf/hadoop-gryo.properties | 3 +
.../spark/groovy/plugin/SparkLoader.groovy | 13 +-
.../groovy/SparkHadoopGremlinPluginTest.java | 33 -----
.../groovy/plugin/SparkGremlinPluginTest.java | 126 +++++++++++++++++++
.../plugin/SparkHadoopGremlinPluginTest.java | 33 +++++
6 files changed, 201 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/docs/src/reference/implementations.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/reference/implementations.asciidoc b/docs/src/reference/implementations.asciidoc
index 1a58ef9..1b7cd92 100644
--- a/docs/src/reference/implementations.asciidoc
+++ b/docs/src/reference/implementations.asciidoc
@@ -1203,12 +1203,37 @@ constructs directly. There is a `gremlin.spark.graphInputRDD` configuration that
InputRDD>`. An `InputRDD` provides a read method that takes a `SparkContext` and returns a graphRDD. Likewise, use
`gremlin.spark.graphOutputRDD` and the respective `OutputRDD`.
+Using Persisted Contexts
+++++++++++++++++++++++++
+
It is possible to persist the graph RDD between jobs within the `SparkContext` (e.g. SparkServer) by leveraging `PersistedOutputRDD`.
Note that `gremlin.spark.persistContext` should be set to `true` or else the persisted RDD will be destroyed when the `SparkContext` closes.
-The persisted RDD is named by the `gremlin.hadoop.outputLocation` configuration (i.e. named in `SparkContext.getPersistedRDDs()`).
-Finally, `PersistedInputRDD` is used with respective `gremlin.hadoop.inputLocation` to retrieve the persisted RDD from the `SparkContext`.
+The persisted RDD is named by the `gremlin.hadoop.outputLocation` configuration.
+Similarly, `PersistedInputRDD` is used with respective `gremlin.hadoop.inputLocation` to retrieve the persisted RDD from the `SparkContext`.
+
+There is a `spark` static object that can be leveraged must like `hdfs` to manage persisted RDDs.
-When using a persistent `Spark Context` the configuration used by the original Spark Configuration will be inherited by all threaded
+[gremlin-groovy]
+----
+spark.create('local[4]')
+graph = GraphFactory.open('conf/hadoop/hadoop-gryo.properties')
+graph.configuration().setProperty('gremlin.spark.persistContext',true)
+graph.configuration().setProperty('gremlin.spark.graphOutputRDD','org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD')
+graph.configuration().setProperty('gremlin.hadoop.outputLocation','pageRankGraph')
+graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().create()).submit().get()
+spark.ls()
+graph.configuration().setProperty('gremlin.hadoop.outputLocation','peerPressureGraph')
+graph.compute(SparkGraphComputer).program(PeerPressureVertexProgram.build().create()).submit().get()
+spark.ls()
+spark.rm('pageRankGraph')
+spark.head('peerPressureGraph')
+spark.describe('peerPressureGraph')
+spark.rm('peerPressureGraph')
+spark.ls()
+spark.close()
+----
+
+When using a persistent `SparkContext` the configuration used by the original Spark Configuration will be inherited by all threaded
references to that Spark Context. The exception to this rule are those properties which have a specific thread local effect.
.Thread Local Properties
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/hadoop-gremlin/conf/hadoop-gryo.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-gryo.properties b/hadoop-gremlin/conf/hadoop-gryo.properties
index 5a43e3e..9bbd41b 100644
--- a/hadoop-gremlin/conf/hadoop-gryo.properties
+++ b/hadoop-gremlin/conf/hadoop-gryo.properties
@@ -28,12 +28,15 @@ gremlin.hadoop.outputLocation=output
spark.master=local[4]
spark.executor.memory=1g
spark.serializer=org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer
+# gremlin.spark.persistContext=true
+# gremlin.spark.graphOutputRDD=org.apache.tinkerpop.gremlin.spark.structure.io.PersistedOutputRDD
# spark.kryo.registrationRequired=true
# spark.storage.memoryFraction=0.2
# spark.eventLog.enabled=true
# spark.eventLog.dir=/tmp/spark-event-logs
# spark.ui.killEnabled=true
+
#####################################
# GiraphGraphComputer Configuration #
#####################################
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy b/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy
index 2f2a095..53d385f 100644
--- a/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy
+++ b/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.spark.groovy.plugin
import org.apache.spark.rdd.RDD
import org.apache.tinkerpop.gremlin.spark.structure.Spark
+import scala.collection.JavaConversions
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -39,14 +40,18 @@ class SparkLoader {
Spark.metaClass.static.rm = { final String rddName ->
for (final RDD<?> rdd : Spark.getRDDs()) {
- if (rdd.name().matches(rddName))
+ if (rdd.name().matches(rddName.replace(".", "\\.").replace("*", ".*")))
Spark.removeRDD(rdd.name());
}
}
+ Spark.metaClass.static.head = { final String rddName ->
+ return Spark.head(rddName, Integer.MAX_VALUE);
+ }
+
Spark.metaClass.static.head = { final String rddName, final int totalLines ->
final List<Object> data = new ArrayList<>();
- final Iterator<?> itty = Spark.getRDD(rddName).toLocalIterator();
+ final Iterator<?> itty = JavaConversions.asJavaIterator(Spark.getRDD(rddName).toLocalIterator());
for (int i = 0; i < totalLines; i++) {
if (itty.hasNext())
data.add(itty.next());
@@ -55,5 +60,9 @@ class SparkLoader {
}
return data;
}
+
+ Spark.metaClass.static.describe = { final String rddName ->
+ return Spark.getRDD(rddName).toDebugString();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkHadoopGremlinPluginTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkHadoopGremlinPluginTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkHadoopGremlinPluginTest.java
deleted file mode 100644
index e412413..0000000
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/SparkHadoopGremlinPluginTest.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.process.computer.groovy;
-
-import org.apache.tinkerpop.gremlin.GraphProviderClass;
-import org.apache.tinkerpop.gremlin.hadoop.groovy.plugin.HadoopPluginSuite;
-import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
-import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
-import org.junit.runner.RunWith;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-@RunWith(HadoopPluginSuite.class)
-@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
-public class SparkHadoopGremlinPluginTest {
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkGremlinPluginTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkGremlinPluginTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkGremlinPluginTest.java
new file mode 100644
index 0000000..932552a
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkGremlinPluginTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.process.computer.groovy.plugin;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.groovy.util.TestableConsolePluginAcceptor;
+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.ranking.pagerank.PageRankVertexProgram;
+import org.apache.tinkerpop.gremlin.spark.AbstractSparkTest;
+import org.apache.tinkerpop.gremlin.spark.groovy.plugin.SparkGremlinPlugin;
+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.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.Before;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class SparkGremlinPluginTest extends AbstractSparkTest {
+
+ @Before
+ public void setupTest() {
+ try {
+ this.console = new TestableConsolePluginAcceptor();
+ final SparkGremlinPlugin plugin = new SparkGremlinPlugin();
+ plugin.pluginTo(this.console);
+ this.console.eval("import " + PageRankVertexProgram.class.getPackage().getName() + ".*");
+ } catch (final Exception e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ ///////////////////
+
+ private TestableConsolePluginAcceptor console;
+
+ @Test
+ public void shouldSupportBasicRDDOperations() throws Exception {
+ String rddName = "target/test-output/graph-1";
+ 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_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_HADOOP_OUTPUT_LOCATION, rddName);
+ configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true); // because the spark context is NOT persisted, neither is the RDD
+ Graph graph = GraphFactory.open(configuration);
+
+ Spark.create("local[4]");
+
+ assertEquals(0, ((List<String>) this.console.eval("spark.ls()")).size());
+
+ this.console.addBinding("graph", graph);
+ this.console.eval("graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().iterations(1).create()).submit().get()");
+ assertEquals(1, ((List<String>) this.console.eval("spark.ls()")).size());
+ assertEquals(rddName + " [Memory Deserialized 1x Replicated]", ((List<String>) this.console.eval("spark.ls()")).get(0));
+
+ rddName = "target/test-output/graph-2";
+ this.console.eval("graph.configuration().setProperty('" + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + "','" + rddName + "')");
+ this.console.eval("graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().iterations(1).create()).submit().get()");
+ assertEquals(2, ((List<String>) this.console.eval("spark.ls()")).size());
+ assertTrue(((List<String>) this.console.eval("spark.ls()")).contains(rddName + " [Memory Deserialized 1x Replicated]"));
+
+ this.console.eval("spark.rm('target/test-output/graph-2')");
+ assertEquals(1, ((List<String>) this.console.eval("spark.ls()")).size());
+ assertTrue(((List<String>) this.console.eval("spark.ls()")).contains("target/test-output/graph-1 [Memory Deserialized 1x Replicated]"));
+
+ assertEquals(6, ((List<Object>) this.console.eval("spark.head('target/test-output/graph-1')")).size());
+
+ this.console.eval("spark.rm('target/test-output/graph-*')");
+ assertEquals(0, ((List<String>) this.console.eval("spark.ls()")).size());
+
+ //////
+ rddName = "target/test-output/graph-1";
+ this.console.eval("graph.configuration().setProperty('" + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + "','" + rddName + "')");
+ this.console.eval("graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().iterations(1).create()).submit().get()");
+
+ rddName = "target/test-output/graph-2";
+ this.console.eval("graph.configuration().setProperty('" + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + "','" + rddName + "')");
+ this.console.eval("graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().iterations(1).create()).submit().get()");
+
+ rddName = "target/test-output/x";
+ this.console.eval("graph.configuration().setProperty('" + Constants.GREMLIN_HADOOP_OUTPUT_LOCATION + "','" + rddName + "')");
+ this.console.eval("graph.compute(SparkGraphComputer).program(PageRankVertexProgram.build().iterations(1).create()).submit().get()");
+
+ assertEquals(3, ((List<String>) this.console.eval("spark.ls()")).size());
+ this.console.eval("spark.rm('target/test-output/graph-*')");
+ assertEquals(1, ((List<String>) this.console.eval("spark.ls()")).size());
+ this.console.eval("spark.rm('*')");
+ assertEquals(0, ((List<String>) this.console.eval("spark.ls()")).size());
+
+ //
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/f8fabe20/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkHadoopGremlinPluginTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkHadoopGremlinPluginTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkHadoopGremlinPluginTest.java
new file mode 100644
index 0000000..63e7b7e
--- /dev/null
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/groovy/plugin/SparkHadoopGremlinPluginTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.process.computer.groovy.plugin;
+
+import org.apache.tinkerpop.gremlin.GraphProviderClass;
+import org.apache.tinkerpop.gremlin.hadoop.groovy.plugin.HadoopPluginSuite;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.spark.process.computer.SparkHadoopGraphProvider;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+@RunWith(HadoopPluginSuite.class)
+@GraphProviderClass(provider = SparkHadoopGraphProvider.class, graph = HadoopGraph.class)
+public class SparkHadoopGremlinPluginTest {
+}