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 16:01:01 UTC
incubator-tinkerpop git commit: fist push on a Spark object for
managing persisted RDDs. Not finished yet.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1023 [created] 9d6467c8f
fist push on a Spark object for managing persisted RDDs. Not finished yet.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/9d6467c8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/9d6467c8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/9d6467c8
Branch: refs/heads/TINKERPOP-1023
Commit: 9d6467c8f48cd34e83270d3f3eabbcce9ce74f05
Parents: 6ae06da
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Dec 8 08:00:47 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Dec 8 08:00:47 2015 -0700
----------------------------------------------------------------------
.../IncidentToAdjacentStrategy.java | 8 +--
.../spark/groovy/plugin/SparkLoader.groovy | 59 ++++++++++++++++++++
.../spark/groovy/plugin/SparkGremlinPlugin.java | 5 +-
.../gremlin/spark/structure/SparkTest.java | 10 ++++
4 files changed, 76 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9d6467c8/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/IncidentToAdjacentStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/IncidentToAdjacentStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/IncidentToAdjacentStrategy.java
index 15327d3..712110d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/IncidentToAdjacentStrategy.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/strategy/optimization/IncidentToAdjacentStrategy.java
@@ -33,6 +33,7 @@ import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.javatuples.Pair;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@@ -62,10 +63,7 @@ public final class IncidentToAdjacentStrategy extends AbstractTraversalStrategy<
implements TraversalStrategy.OptimizationStrategy {
private static final IncidentToAdjacentStrategy INSTANCE = new IncidentToAdjacentStrategy();
- private static final Set<Class> InvalidatingStepClasses = new HashSet<Class>() {{
- add(PathStep.class);
- add(LambdaHolder.class);
- }};
+ private static final Set<Class> INVALIDATING_STEP_CLASSES = new HashSet<>(Arrays.asList(PathStep.class, LambdaHolder.class));
private IncidentToAdjacentStrategy() {
}
@@ -73,7 +71,7 @@ public final class IncidentToAdjacentStrategy extends AbstractTraversalStrategy<
@Override
public void apply(Traversal.Admin<?, ?> traversal) {
final Traversal.Admin root = TraversalHelper.getRootTraversal(traversal);
- if (TraversalHelper.hasStepOfAssignableClassRecursively(InvalidatingStepClasses, root))
+ if (TraversalHelper.hasStepOfAssignableClassRecursively(INVALIDATING_STEP_CLASSES, root))
return;
final Collection<Pair<VertexStep, Step>> stepsToReplace = new ArrayList<>();
Step prev = null;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9d6467c8/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
new file mode 100644
index 0000000..2f2a095
--- /dev/null
+++ b/spark-gremlin/src/main/groovy/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkLoader.groovy
@@ -0,0 +1,59 @@
+/*
+ * 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.groovy.plugin
+
+import org.apache.spark.rdd.RDD
+import org.apache.tinkerpop.gremlin.spark.structure.Spark
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+class SparkLoader {
+
+ public static void load() {
+
+ Spark.metaClass.static.ls = {
+ final List<String> rdds = new ArrayList<>();
+ for (final RDD<?> rdd : Spark.getRDDs()) {
+ rdds.add(rdd.name() + " [" + rdd.getStorageLevel().description() + "]")
+ }
+ return rdds;
+ }
+
+ Spark.metaClass.static.rm = { final String rddName ->
+ for (final RDD<?> rdd : Spark.getRDDs()) {
+ if (rdd.name().matches(rddName))
+ Spark.removeRDD(rdd.name());
+ }
+ }
+
+ Spark.metaClass.static.head = { final String rddName, final int totalLines ->
+ final List<Object> data = new ArrayList<>();
+ final Iterator<?> itty = Spark.getRDD(rddName).toLocalIterator();
+ for (int i = 0; i < totalLines; i++) {
+ if (itty.hasNext())
+ data.add(itty.next());
+ else
+ break;
+ }
+ return data;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9d6467c8/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkGremlinPlugin.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkGremlinPlugin.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkGremlinPlugin.java
index af6d421..9351a1e 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkGremlinPlugin.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/SparkGremlinPlugin.java
@@ -26,7 +26,7 @@ import org.apache.tinkerpop.gremlin.groovy.plugin.PluginAcceptor;
import org.apache.tinkerpop.gremlin.groovy.plugin.PluginInitializationException;
import org.apache.tinkerpop.gremlin.groovy.plugin.RemoteAcceptor;
import org.apache.tinkerpop.gremlin.spark.process.computer.SparkGraphComputer;
-import org.slf4j.Logger;
+import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import java.util.HashSet;
import java.util.Optional;
@@ -42,6 +42,7 @@ public final class SparkGremlinPlugin extends AbstractGremlinPlugin {
protected static final Set<String> IMPORTS = new HashSet<String>() {{
add("import org.apache.log4j.*");
add(IMPORT_SPACE + SparkGraphComputer.class.getPackage().getName() + DOT_STAR);
+ add(IMPORT_SPACE + Spark.class.getPackage().getName() + DOT_STAR);
}};
@Override
@@ -55,6 +56,8 @@ public final class SparkGremlinPlugin extends AbstractGremlinPlugin {
try {
pluginAcceptor.eval(String.format("Logger.getLogger(%s).setLevel(Level.INFO)", SparkGraphComputer.class.getName()));
pluginAcceptor.eval(String.format("Logger.getLogger(%s).setLevel(Level.ERROR)", MetricsSystem.class.getName()));
+ pluginAcceptor.eval("spark = Spark");
+ pluginAcceptor.eval(SparkLoader.class.getCanonicalName() + ".load()");
} catch (final Exception e) {
throw new PluginInitializationException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/9d6467c8/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
index 9e9882c..e5cc471 100644
--- 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
@@ -69,6 +69,16 @@ public class SparkTest extends AbstractSparkTest {
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());
+
+ System.out.println(Spark.getRDD(prefix + i).getStorageLevel().description());
+ System.out.println(Spark.getRDD(prefix + i).getCreationSite());
+ System.out.println(Spark.getRDD(prefix + i).id());
+ System.out.println(Spark.getRDD(prefix + i).toDebugString());
+ System.out.println(Spark.getRDD(prefix + i).logName());
+ if(true) {
+ throw new IllegalArgumentException("HAA");
+ }
+
}
for (int i = 9; i >= 0; i--) {