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/04/22 00:09:21 UTC
incubator-tinkerpop git commit: SparkExecutor is smart about vertex
property and edge dropping with StarGraph.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 24ffbddfe -> 7991b9173
SparkExecutor is smart about vertex property and edge dropping with StarGraph.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/7991b917
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/7991b917
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/7991b917
Branch: refs/heads/master
Commit: 7991b917305a51d0aa5467aec0db038c185db5e0
Parents: 24ffbdd
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Apr 21 16:09:17 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Apr 21 16:09:17 2015 -0600
----------------------------------------------------------------------
.../gremlin/structure/util/star/StarGraph.java | 27 ++++++-------------
.../process/computer/spark/SparkExecutor.java | 28 ++++++++++----------
.../computer/spark/SparkGraphComputer.java | 1 -
3 files changed, 22 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/7991b917/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
index 1e47e6d..d4e785e 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/star/StarGraph.java
@@ -33,7 +33,6 @@ import org.apache.tinkerpop.gremlin.structure.T;
import org.apache.tinkerpop.gremlin.structure.Transaction;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
@@ -62,7 +61,7 @@ public final class StarGraph implements Graph, Serializable {
private StarVertex starVertex = null;
private Long nextId = 0l;
- private final Map<Object, Map<String, Object>> edgeProperties = new HashMap<>();
+ private final Map<Object, Map<String, Object>> edgeProperties = new HashMap<>(); // TODO: null these until required? (space savings)
private final Map<Object, Map<String, Object>> metaProperties = new HashMap<>();
private StarGraph() {
@@ -236,11 +235,11 @@ public final class StarGraph implements Graph, Serializable {
//// STAR VERTEX ////
/////////////////////
- public final class StarVertex extends StarElement implements Vertex, Attachable<Vertex> {
+ public final class StarVertex extends StarElement implements Vertex {
- private Map<String, List<Edge>> outEdges = new HashMap<>();
- private Map<String, List<Edge>> inEdges = new HashMap<>();
- private Map<String, List<VertexProperty>> vertexProperties = new HashMap<>();
+ private final Map<String, List<Edge>> outEdges = new HashMap<>();
+ private final Map<String, List<Edge>> inEdges = new HashMap<>();
+ private final Map<String, List<VertexProperty>> vertexProperties = new HashMap<>();
public StarVertex(final Object id, final String label) {
super(id, label);
@@ -250,24 +249,14 @@ public final class StarGraph implements Graph, Serializable {
return this;
}
- public Vertex attach(final Vertex hostVertex, final Method method) {
- return (Vertex) method.apply(this, hostVertex);
- }
-
- public Vertex attach(final Graph hostGraph, final Method method) {
- return (Vertex) method.apply(this, hostGraph);
- }
-
public void dropEdges() {
this.outEdges.clear();
this.inEdges.clear();
}
- public void dropVertexProperties(final String... keys) {
- if (null != keys) { // TODO: this is bad
- for (final String key : keys) {
- this.vertexProperties.remove(key);
- }
+ public void dropVertexProperties(final String... propertyKeys) {
+ for (final String key : propertyKeys) {
+ this.vertexProperties.remove(key);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/7991b917/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
index af0a6a6..485a953 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
@@ -41,12 +41,11 @@ import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.util.Attachable;
-import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.util.Attachable;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
+import org.apache.tinkerpop.gremlin.structure.util.star.StarGraph;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import scala.Tuple2;
@@ -60,6 +59,8 @@ import java.util.Set;
*/
public final class SparkExecutor {
+ private static final String[] EMPTY_ARRAY = new String[0];
+
private SparkExecutor() {
}
@@ -75,18 +76,18 @@ public final class SparkExecutor {
final JavaPairRDD<Object, ViewOutgoingPayload<M>> viewOutgoingRDD = (((null == viewIncomingRDD) ?
graphRDD.mapValues(vertexWritable -> new Tuple2<>(vertexWritable, Optional.<ViewIncomingPayload<M>>absent())) : // first iteration will not have any views or messages
- graphRDD.leftOuterJoin(viewIncomingRDD)) // every other iteration may have views and messages
+ graphRDD.leftOuterJoin(viewIncomingRDD)) // every other iteration may have views and messages
// for each partition of vertices
.mapPartitionsToPair(partitionIterator -> {
final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration); // each partition(Spark)/worker(TP3) has a local copy of the vertex program (a worker's task)
final Set<String> elementComputeKeys = workerVertexProgram.getElementComputeKeys(); // the compute keys as a set
- final String[] elementComputeKeysArray = elementComputeKeys.size() == 0 ? null : elementComputeKeys.toArray(new String[elementComputeKeys.size()]); // the compute keys as an array
+ final String[] elementComputeKeysArray = elementComputeKeys.size() == 0 ? EMPTY_ARRAY : elementComputeKeys.toArray(new String[elementComputeKeys.size()]); // the compute keys as an array
final SparkMessenger<M> messenger = new SparkMessenger<>();
workerVertexProgram.workerIterationStart(memory); // start the worker
return () -> IteratorUtils.map(partitionIterator, vertexViewIncoming -> {
final Vertex vertex = vertexViewIncoming._2()._1().get(); // get the vertex from the vertex writable
- // drop any compute properties that are in memory
- ((StarGraph.StarVertex)vertex).dropVertexProperties(elementComputeKeysArray);
+ // drop any compute properties that are cached in memory
+ if (elementComputeKeysArray.length > 0) ((StarGraph.StarVertex) vertex).dropVertexProperties(elementComputeKeysArray);
final boolean hasViewAndMessages = vertexViewIncoming._2()._2().isPresent(); // if this is the first iteration, then there are no views or messages
final List<DetachedVertexProperty<Object>> previousView = hasViewAndMessages ? vertexViewIncoming._2()._2().get().getView() : Collections.emptyList();
final List<M> incomingMessages = hasViewAndMessages ? vertexViewIncoming._2()._2().get().getIncomingMessages() : Collections.emptyList();
@@ -95,7 +96,7 @@ public final class SparkExecutor {
messenger.setVertexAndIncomingMessages(vertex, incomingMessages); // set the messenger with the incoming messages
workerVertexProgram.execute(ComputerGraph.of(vertex, elementComputeKeys), messenger, memory); // execute the vertex program on this vertex for this iteration
///
- final List<DetachedVertexProperty<Object>> nextView = null == elementComputeKeysArray ? // not all vertex programs have compute keys
+ final List<DetachedVertexProperty<Object>> nextView = elementComputeKeysArray.length == 0 ? // not all vertex programs have compute keys
Collections.emptyList() :
IteratorUtils.list(IteratorUtils.map(vertex.properties(elementComputeKeysArray), property -> DetachedFactory.detach(property, true)));
final List<Tuple2<Object, M>> outgoingMessages = messenger.getOutgoingMessages(); // get the outgoing messages
@@ -126,7 +127,7 @@ public final class SparkExecutor {
}
})
.filter(payload -> !(payload._2() instanceof MessagePayload)) // this happens if there is a message to a vertex that does not exist
- .filter(payload -> !((payload._2() instanceof ViewIncomingPayload) && !((ViewIncomingPayload<M>)payload._2()).hasView())) // this happens if there are many messages to a vertex that does not exist
+ .filter(payload -> !((payload._2() instanceof ViewIncomingPayload) && !((ViewIncomingPayload<M>) payload._2()).hasView())) // this happens if there are many messages to a vertex that does not exist
.mapValues(payload -> payload instanceof ViewIncomingPayload ?
(ViewIncomingPayload<M>) payload : // this happens if there is a vertex with incoming messages
new ViewIncomingPayload<>((ViewPayload) payload)); // this happens if there is a vertex with no incoming messages
@@ -142,17 +143,16 @@ public final class SparkExecutor {
////////////////
public static <M> JavaPairRDD<Object, VertexWritable> prepareGraphRDDForMapReduce(final JavaPairRDD<Object, VertexWritable> graphRDD, final JavaPairRDD<Object, ViewIncomingPayload<M>> viewIncomingRDD, final String[] elementComputeKeys) {
- return (null == viewIncomingRDD) ?
+ return (null == viewIncomingRDD) ? // there was no vertex program
graphRDD.mapValues(vertexWritable -> {
- ((StarGraph.StarVertex)vertexWritable.get()).dropEdges();
- ((StarGraph.StarVertex)vertexWritable.get()).dropVertexProperties(elementComputeKeys);
+ ((StarGraph.StarVertex) vertexWritable.get()).dropEdges();
return vertexWritable;
}) :
graphRDD.leftOuterJoin(viewIncomingRDD)
.mapValues(tuple -> {
final Vertex vertex = tuple._1().get();
- ((StarGraph.StarVertex)vertex).dropEdges();
- ((StarGraph.StarVertex)vertex).dropVertexProperties(elementComputeKeys);
+ ((StarGraph.StarVertex) vertex).dropEdges();
+ ((StarGraph.StarVertex) vertex).dropVertexProperties(elementComputeKeys);
final List<DetachedVertexProperty<Object>> view = tuple._2().isPresent() ? tuple._2().get().getView() : Collections.emptyList();
view.forEach(property -> property.attach(vertex, Attachable.Method.CREATE));
return tuple._1();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/7991b917/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
index d40b932..09e528e 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkGraphComputer.java
@@ -213,7 +213,6 @@ public final class SparkGraphComputer implements GraphComputer {
// process the map reducers //
//////////////////////////////
if (!this.mapReducers.isEmpty()) {
- // drop all edges and messages in the graphRDD as they are no longer needed for the map reduce jobs
final String[] elementComputeKeys = this.vertexProgram == null ? new String[0] : this.vertexProgram.getElementComputeKeys().toArray(new String[this.vertexProgram.getElementComputeKeys().size()]);
final JavaPairRDD<Object, VertexWritable> mapReduceGraphRDD = SparkExecutor.prepareGraphRDDForMapReduce(graphRDD, viewIncomingRDD, elementComputeKeys).setName("mapReduceGraphRDD").cache();
for (final MapReduce mapReduce : this.mapReducers) {