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/03/25 19:09:22 UTC
incubator-tinkerpop git commit: There is now Persist and ResultGraph
semantics in GraphComputer. VertexPrograms like TraversalVertexProgram have
Persist.NOTHING and ResultGraph.ORIGINAL_GRAPH. This greatly reduces the data
footprint and the output write
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master f590faacc -> 5a4428f76
There is now Persist and ResultGraph semantics in GraphComputer. VertexPrograms like TraversalVertexProgram have Persist.NOTHING and ResultGraph.ORIGINAL_GRAPH. This greatly reduces the data footprint and the output write speed of the Graph in, for example, Hadoop settings. This also makes it such that vendors can now support writing back results of the computed results to the original graph (e.g. calculate pagerank and store it back in the graph).
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/5a4428f7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/5a4428f7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/5a4428f7
Branch: refs/heads/master
Commit: 5a4428f765664a54e315b61f5b8e58260b036799
Parents: f590faa
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Mar 25 12:09:15 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Mar 25 12:09:15 2015 -0600
----------------------------------------------------------------------
.gitignore | 1 +
gremlin-core/pom.xml | 2 +-
.../gremlin/process/computer/GraphComputer.java | 43 ++++++++++-
.../gremlin/process/computer/VertexProgram.java | 5 ++
.../peerpressure/PeerPressureVertexProgram.java | 16 +++-
.../computer/lambda/LambdaVertexProgram.java | 13 +++-
.../ranking/pagerank/PageRankVertexProgram.java | 14 +++-
.../traversal/TraversalVertexProgram.java | 27 +++++--
.../traversal/step/map/ComputerResultStep.java | 2 +-
.../process/computer/GraphComputerTest.java | 14 +++-
.../ranking/PageRankVertexProgramTest.java | 2 +-
.../structure/ExceptionCoverageTest.java | 2 +-
.../tinkerpop/gremlin/hadoop/Constants.java | 14 ++--
.../computer/giraph/GiraphComputeVertex.java | 2 +-
.../computer/giraph/GiraphGraphComputer.java | 77 +++++++++++++++-----
.../process/computer/giraph/GiraphMemory.java | 8 +-
.../computer/spark/SparkGraphComputer.java | 38 ++++++++--
.../computer/spark/util/SparkHelper.java | 4 +-
.../process/computer/util/MemoryMapReduce.java | 6 +-
.../io/graphson/GraphSONOutputFormat.java | 4 +-
.../io/graphson/GraphSONRecordReader.java | 9 ++-
.../io/graphson/GraphSONRecordWriter.java | 30 +++++---
.../structure/io/gryo/GryoOutputFormat.java | 4 +-
.../structure/io/gryo/GryoRecordReader.java | 7 +-
.../structure/io/gryo/GryoRecordWriter.java | 25 ++++---
.../hadoop/structure/util/HadoopHelper.java | 8 +-
.../gremlin/hadoop/HadoopGraphProvider.java | 3 +-
.../GraphSONRecordReaderWriterTest.java | 12 +--
.../io/gryo/GryoRecordReaderWriterTest.java | 10 +--
pom.xml | 1 +
.../process/computer/TinkerGraphComputer.java | 29 +++++++-
31 files changed, 329 insertions(+), 103 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index f4779c5..9a072bb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,3 +14,4 @@ _bsp/
doc/out
docs/*.asciidoc
ext/
+dependency-reduced-pom.xml
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-core/pom.xml b/gremlin-core/pom.xml
index e449b1a..4ba209a 100644
--- a/gremlin-core/pom.xml
+++ b/gremlin-core/pom.xml
@@ -133,7 +133,7 @@ limitations under the License.
<goal>shade</goal>
</goals>
<configuration>
- <createDependencyReducedPom>false</createDependencyReducedPom>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
<artifactSet>
<includes>
<include>com.esotericsoftware.kryo:*</include>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
index 9bbde12..c1f9a9f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputer.java
@@ -44,6 +44,32 @@ public interface GraphComputer {
DIRTY_BSP
}
+ public enum ResultGraph {
+ /**
+ * When the computation is complete, the {@link org.apache.tinkerpop.gremlin.structure.Graph} in {@link ComputerResult} is the original graph that spawned the graph computer.
+ */
+ ORIGINAL_GRAPH,
+ /**
+ * When the computation is complete, the {@link org.apache.tinkerpop.gremlin.structure.Graph} in {@link ComputerResult} is a new graph cloned from the original graph.
+ */
+ NEW_GRAPH
+ }
+
+ public enum Persist {
+ /**
+ * Write nothing to the declared {@link ResultGraph}.
+ */
+ NOTHING,
+ /**
+ * Write vertex and vertex properties back to the {@link ResultGraph}.
+ */
+ VERTEX_PROPERTIES,
+ /**
+ * Write vertex, vertex properties, and edges back to the {@link ResultGraph}.
+ */
+ EDGES
+ }
+
/**
* Set the {@link Isolation} of the computation.
*
@@ -53,6 +79,17 @@ public interface GraphComputer {
public GraphComputer isolation(final Isolation isolation);
/**
+ * Set the {@link ResultGraph} of the computation.
+ * If this is not set explicitly by the user, then the {@link VertexProgram} can choose the most efficient result for its intended use.
+ *
+ * @param resultGraph the type of graph to be returned by {@link ComputerResult#graph}
+ * @return the updated GraphComputer with newly set result graph.
+ */
+ public GraphComputer result(final ResultGraph resultGraph);
+
+ public GraphComputer persist(final Persist persist);
+
+ /**
* Set the {@link VertexProgram} to be executed by the {@link GraphComputer}.
* There can only be one VertexProgram for the GraphComputer.
*
@@ -131,7 +168,7 @@ public interface GraphComputer {
return true;
}
- public default boolean supportsNonSerializableObjects() {
+ public default boolean supportsDirectObjects() {
return true;
}
}
@@ -161,6 +198,10 @@ public interface GraphComputer {
return new IllegalArgumentException("The provided isolation is not supported by this graph computer: " + isolation);
}
+ public static IllegalArgumentException resultGraphPersistCombinationNotSupported(final ResultGraph resultGraph, final Persist persist) {
+ return new IllegalArgumentException("The computer does not support the following result graph and persist combination: " + resultGraph + ":" + persist);
+ }
+
public static IllegalStateException computerHasAlreadyBeenSubmittedAVertexProgram() {
return new IllegalStateException("This computer has already had a vertex program submitted to it");
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
index 4cfebb2..06f5f4d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
@@ -195,6 +195,11 @@ public interface VertexProgram<M> extends Cloneable {
@SuppressWarnings("CloneDoesntDeclareCloneNotSupportedException")
public VertexProgram<M> clone();
+ public GraphComputer.ResultGraph getPreferredResultGraph();
+
+ public GraphComputer.Persist getPreferredPersist();
+
+
/**
* A helper method to construct a {@link VertexProgram} given the content of the supplied configuration.
* The class of the VertexProgram is read from the {@link VertexProgram#VERTEX_PROGRAM} static configuration key.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
index ca1c0a6..f0899fc 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
@@ -18,21 +18,21 @@
*/
package org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.LambdaHolder;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticVertexProgram;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.MapHelper;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.StreamFactory;
-import org.apache.commons.configuration.Configuration;
import org.javatuples.Pair;
import java.io.Serializable;
@@ -109,6 +109,16 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
}
@Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.NEW_GRAPH;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.VERTEX_PROPERTIES;
+ }
+
+ @Override
public void setup(final Memory memory) {
memory.set(VOTE_TO_HALT, false);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/lambda/LambdaVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/lambda/LambdaVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/lambda/LambdaVertexProgram.java
index a971588..a3b06f6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/lambda/LambdaVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/lambda/LambdaVertexProgram.java
@@ -18,6 +18,8 @@
*/
package org.apache.tinkerpop.gremlin.process.computer.lambda;
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
@@ -28,7 +30,6 @@ import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.function.TriConsumer;
-import org.apache.commons.configuration.Configuration;
import java.io.Serializable;
import java.util.Arrays;
@@ -134,6 +135,16 @@ public class LambdaVertexProgram<M extends Serializable> extends StaticVertexPro
}
@Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.ORIGINAL_GRAPH;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.NOTHING;
+ }
+
+ @Override
public String toString() {
return StringFactory.vertexProgramString(this);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
index 774a476..de88430 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank;
import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
@@ -27,10 +27,10 @@ import org.apache.tinkerpop.gremlin.process.computer.Messenger;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.LambdaHolder;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticVertexProgram;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.StreamFactory;
@@ -91,6 +91,16 @@ public class PageRankVertexProgram extends StaticVertexProgram<Double> {
}
@Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.NEW_GRAPH;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.VERTEX_PROPERTIES;
+ }
+
+ @Override
public Set<String> getElementComputeKeys() {
return COMPUTE_KEYS;
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index bbd221d..97616d6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -19,12 +19,8 @@
package org.apache.tinkerpop.gremlin.process.computer.traversal;
import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.TraverserGenerator;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
@@ -35,13 +31,18 @@ import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.Computer
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.sideEffect.mapreduce.TraverserMapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.LambdaHolder;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.TraverserGenerator;
+import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.GraphStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SideEffectCapStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
-import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Element;
@@ -229,6 +230,16 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
}
@Override
+ public GraphComputer.ResultGraph getPreferredResultGraph() {
+ return GraphComputer.ResultGraph.ORIGINAL_GRAPH;
+ }
+
+ @Override
+ public GraphComputer.Persist getPreferredPersist() {
+ return GraphComputer.Persist.NOTHING;
+ }
+
+ @Override
public String toString() {
final String traversalString = this.traversal.toString().substring(1);
return StringFactory.vertexProgramString(this, traversalString.substring(0, traversalString.length() - 1));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
index 1f9636e..3b68844 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
@@ -75,7 +75,7 @@ public final class ComputerResultStep<S> extends AbstractStep<S, S> {
if (this.byPass) return this.starts.next();
if (this.first && null == this.computerResult) {
try {
- populateTraversers(this.graphComputer.program(TraversalVertexProgram.build().traversal(this.getTraversal(), !this.graphComputer.features().supportsNonSerializableObjects()).create()).submit().get());
+ populateTraversers(this.graphComputer.program(TraversalVertexProgram.build().traversal(this.getTraversal(), !this.graphComputer.features().supportsDirectObjects()).create()).submit().get());
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
index 895a0dd..778b998 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/GraphComputerTest.java
@@ -52,7 +52,8 @@ import static org.junit.Assert.*;
"computerHasAlreadyBeenSubmittedAVertexProgram",
"providedKeyIsNotAnElementComputeKey",
"isolationNotSupported",
- "incidentAndAdjacentElementsCanNotBeAccessedInMapReduce"
+ "incidentAndAdjacentElementsCanNotBeAccessedInMapReduce",
+ "resultGraphPersistCombinationNotSupported" // TODO: NOT TRUE!
})
@ExceptionCoverage(exceptionClass = Graph.Exceptions.class, methods = {
"graphDoesNotSupportProvidedGraphComputer",
@@ -615,6 +616,16 @@ public abstract class GraphComputerTest extends AbstractGremlinProcessTest {
}
@Override
+ public GraphComputer result(final ResultGraph resultGraph) {
+ return null;
+ }
+
+ @Override
+ public GraphComputer persist(final Persist persist) {
+ return null;
+ }
+
+ @Override
public GraphComputer program(final VertexProgram vertexProgram) {
return null;
}
@@ -628,5 +639,6 @@ public abstract class GraphComputerTest extends AbstractGremlinProcessTest {
public Future<ComputerResult> submit() {
return null;
}
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
index b7159f4..51734e1 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
@@ -41,7 +41,7 @@ public class PageRankVertexProgramTest extends AbstractGremlinProcessTest {
final ComputerResult result = graph.compute().program(PageRankVertexProgram.build().create()).submit().get();
result.graph().traversal().V().forEachRemaining(v -> {
assertTrue(v.keys().contains("name"));
- assertFalse(v.keys().contains(PageRankVertexProgram.PAGE_RANK));
+ assertTrue(v.keys().contains(PageRankVertexProgram.PAGE_RANK));
final String name = v.value("name");
final Double pageRank = v.value(PageRankVertexProgram.PAGE_RANK);
//System.out.println(name + "-----" + pageRank);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/gremlin-test/src/test/java/org/apache/tinkerpop/gremlin/structure/ExceptionCoverageTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/test/java/org/apache/tinkerpop/gremlin/structure/ExceptionCoverageTest.java b/gremlin-test/src/test/java/org/apache/tinkerpop/gremlin/structure/ExceptionCoverageTest.java
index b8a417f..e0cef89 100644
--- a/gremlin-test/src/test/java/org/apache/tinkerpop/gremlin/structure/ExceptionCoverageTest.java
+++ b/gremlin-test/src/test/java/org/apache/tinkerpop/gremlin/structure/ExceptionCoverageTest.java
@@ -69,7 +69,7 @@ public class ExceptionCoverageTest {
add("org.apache.tinkerpop.gremlin.process.computer.GraphComputer$Exceptions#adjacentElementPropertiesCanNotBeRead");
add("org.apache.tinkerpop.gremlin.process.computer.GraphComputer$Exceptions#adjacentElementPropertiesCanNotBeWritten");
add("org.apache.tinkerpop.gremlin.process.computer.GraphComputer$Exceptions#adjacentVerticesCanNotBeQueried");
- add("org.apache.tinkerpop.gremlin.process.computer.GraphComputer$Exceptions#supportsNonSerializableObjects");
+ add("org.apache.tinkerpop.gremlin.process.computer.GraphComputer$Exceptions#supportsDirectObjects");
}};
// register test classes here that contain @ExceptionCoverage annotations
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
index 5062bb4..2d70bdc 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/Constants.java
@@ -29,25 +29,27 @@ public class Constants {
public static final String GREMLIN_HADOOP_OUTPUT_LOCATION = "gremlin.hadoop.outputLocation";
public static final String GREMLIN_HADOOP_GRAPH_INPUT_FORMAT = "gremlin.hadoop.graphInputFormat";
public static final String GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT = "gremlin.hadoop.graphOutputFormat";
+ public static final String GREMLIN_HADOOP_GRAPH_INPUT_FORMAT_HAS_EDGES = "gremlin.hadoop.graphOutputFormat.hasEdges";
+ public static final String GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES = "gremlin.hadoop.graphInputFormat.hasEdges";
public static final String GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT = "gremlin.hadoop.memoryOutputFormat";
- public static final String GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER = "gremlin.hadoop.defaultGraphComputer";
+ // public static final String GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER = "gremlin.hadoop.defaultGraphComputer";
public static final String GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE = "gremlin.hadoop.jarsInDistributedCache";
public static final String HIDDEN_G = Graph.Hidden.hide("g");
public static final String GREMLIN_HADOOP_JOB_PREFIX = "HadoopGremlin: ";
public static final String GREMLIN_HADOOP_GIRAPH_JOB_PREFIX = "HadoopGremlin(Giraph): ";
- public static final String GREMLIN_HADOOP_MAP_REDUCE_JOB_PREFIX = "HadoopGremlin(MapReduce): ";
+ // public static final String GREMLIN_HADOOP_MAP_REDUCE_JOB_PREFIX = "HadoopGremlin(MapReduce): ";
public static final String GREMLIN_HADOOP_SPARK_JOB_PREFIX = "HadoopGremlin(Spark): ";
public static final String HADOOP_GREMLIN_LIBS = "HADOOP_GREMLIN_LIBS";
public static final String DOT_JAR = ".jar";
public static final String GREMLIN_HADOOP_DERIVE_MEMORY = "gremlin.hadoop.deriveMemory";
- public static final String SYSTEM_MEMORY = Graph.Hidden.hide("memory");
- public static final String SYSTEM_RUNTIME = Graph.Hidden.hide("gremlin.hadoop.runtime");
- public static final String SYSTEM_ITERATION = Graph.Hidden.hide("gremlin.hadoop.iteration");
+ public static final String HIDDEN_MEMORY = Graph.Hidden.hide("memory"); // TODO: Why not gremlin.hadoop?
+ public static final String HIDDEN_RUNTIME = Graph.Hidden.hide("gremlin.hadoop.runtime");
+ public static final String HIDDEN_ITERATION = Graph.Hidden.hide("gremlin.hadoop.iteration");
public static final String GREMLIN_HADOOP_MEMORY_KEYS = "gremlin.hadoop.memoryKeys";
public static final String GREMLIN_HADOOP_MAP_REDUCE_CLASS = "gremlin.hadoop.mapReduceClass";
public static final String GREMLIN_HADOOP_HALT = "gremlin.hadoop.halt";
- public static final String MAP_MEMORY = "gremlin.hadoop.mapMemory";
+ public static final String GREMLIN_HADOOP_MAP_MEMORY = "gremlin.hadoop.mapMemory";
public static final String MAPRED_INPUT_DIR = "mapred.input.dir";
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
index 4ce5056..3cd32d5 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphComputeVertex.java
@@ -60,7 +60,7 @@ public final class GiraphComputeVertex extends Vertex<ObjectWritable, VertexWrit
final MapMemory mapMemory = new MapMemory();
memory.asMap().forEach(mapMemory::set);
mapMemory.setIteration(memory.getIteration() - 1);
- wrappedVertex.property(VertexProperty.Cardinality.single, Constants.MAP_MEMORY, mapMemory); // TODO: this is a "computer key"
+ wrappedVertex.property(VertexProperty.Cardinality.single, Constants.GREMLIN_HADOOP_MAP_MEMORY, mapMemory); // TODO: this is a "computer key"
}
workerContext.getVertexProgramPool().offer(vertexProgram);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
index a84a7af..fc958d5 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphGraphComputer.java
@@ -31,6 +31,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
@@ -47,17 +48,16 @@ import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
-import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.HashSet;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
@@ -74,6 +74,9 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
protected GiraphConfiguration giraphConfiguration = new GiraphConfiguration();
private boolean executed = false;
+ private Optional<ResultGraph> resultGraph = Optional.empty();
+ private Optional<Persist> persist = Optional.empty();
+
private final Set<MapReduce> mapReduces = new HashSet<>();
private VertexProgram<?> vertexProgram;
private MapMemory memory = new MapMemory();
@@ -101,6 +104,18 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
}
@Override
+ public GraphComputer result(final ResultGraph resultGraph) {
+ this.resultGraph = Optional.of(resultGraph);
+ return this;
+ }
+
+ @Override
+ public GraphComputer persist(final Persist persist) {
+ this.persist = Optional.of(persist);
+ return this;
+ }
+
+ @Override
public GraphComputer program(final VertexProgram vertexProgram) {
this.vertexProgram = vertexProgram;
this.memory.addVertexProgramMemoryComputeKeys(this.vertexProgram);
@@ -147,42 +162,66 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
throw new IllegalStateException(e.getMessage(), e);
}
this.memory.setRuntime(System.currentTimeMillis() - startTime);
-
- final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram);
- return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : new ComputerGraph(outputGraph, this.vertexProgram.getElementComputeKeys()), this.memory.asImmutable());
+ return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph, this.resultGraph.get(), this.persist.get()), this.memory.asImmutable());
});
}
@Override
public int run(final String[] args) {
+
+ if (!this.persist.isPresent())
+ this.persist = Optional.of(null == this.vertexProgram ? Persist.NOTHING : this.vertexProgram.getPreferredPersist());
+ if (!this.resultGraph.isPresent())
+ this.resultGraph = Optional.of(null == this.vertexProgram ? ResultGraph.ORIGINAL_GRAPH : this.vertexProgram.getPreferredResultGraph());
+ if (this.resultGraph.get().equals(ResultGraph.ORIGINAL_GRAPH))
+ if (!this.persist.get().equals(Persist.NOTHING))
+ throw GraphComputer.Exceptions.resultGraphPersistCombinationNotSupported(this.resultGraph.get(), this.persist.get());
+ this.giraphConfiguration.setBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, this.persist.get().equals(Persist.EDGES));
+
try {
// it is possible to run graph computer without a vertex program (and thus, only map reduce jobs if they exist)
if (null != this.vertexProgram) {
- final GiraphJob job = new GiraphJob(this.giraphConfiguration, Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
- final Path inputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
- if (!FileSystem.get(this.giraphConfiguration).exists(inputPath))
- throw new IllegalArgumentException("The provided input path does not exist: " + inputPath);
- FileInputFormat.setInputPaths(job.getInternalJob(), inputPath);
- FileOutputFormat.setOutputPath(job.getInternalJob(), new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_G));
- // job.getInternalJob().setJarByClass(GiraphGraphComputer.class);
- LOGGER.info(Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
- if (!job.run(true)) {
- throw new IllegalStateException("The GiraphGraphComputer job failed -- aborting all subsequent MapReduce jobs");
- }
+ // determine all the requisite map reduce jobs
this.mapReduces.addAll(this.vertexProgram.getMapReducers());
// calculate main vertex program memory if desired (costs one mapreduce job)
if (this.giraphConfiguration.getBoolean(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, false)) {
- final Set<String> memoryKeys = new HashSet<String>(this.vertexProgram.getMemoryComputeKeys());
- memoryKeys.add(Constants.SYSTEM_ITERATION);
+ final Set<String> memoryKeys = new HashSet<>(this.vertexProgram.getMemoryComputeKeys());
+ memoryKeys.add(Constants.HIDDEN_ITERATION);
this.giraphConfiguration.setStrings(Constants.GREMLIN_HADOOP_MEMORY_KEYS, (String[]) memoryKeys.toArray(new String[memoryKeys.size()]));
this.mapReduces.add(new MemoryMapReduce(memoryKeys));
}
+
+ // prepare the giraph vertex-centric computing job
+ final GiraphJob job = new GiraphJob(this.giraphConfiguration, Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
+ final Path inputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
+ final Path outputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_G);
+ if (!FileSystem.get(this.giraphConfiguration).exists(inputPath)) // TODO: what about when the input is not a file input?
+ throw new IllegalArgumentException("The provided input path does not exist: " + inputPath);
+ FileInputFormat.setInputPaths(job.getInternalJob(), inputPath);
+ if (this.persist.get().equals(Persist.NOTHING) && this.mapReduces.isEmpty()) // do not write the graph back if it is not needed in MapReduce
+ job.getInternalJob().setOutputFormatClass(NullOutputFormat.class);
+ else
+ FileOutputFormat.setOutputPath(job.getInternalJob(), outputPath);
+ job.getInternalJob().setJarByClass(GiraphGraphComputer.class);
+ LOGGER.info(Constants.GREMLIN_HADOOP_GIRAPH_JOB_PREFIX + this.vertexProgram);
+ // execute the job and wait until it completes (if it fails, throw an exception)
+ if (!job.run(true))
+ throw new IllegalStateException("The GiraphGraphComputer job failed -- aborting all subsequent MapReduce jobs");
+
}
// do map reduce jobs
+ this.giraphConfiguration.setBoolean(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT_HAS_EDGES, this.giraphConfiguration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true));
for (final MapReduce mapReduce : this.mapReduces) {
this.memory.addMapReduceMemoryKey(mapReduce);
MapReduceHelper.executeMapReduceJob(mapReduce, this.memory, this.giraphConfiguration);
}
+ // if no persistence, delete the map reduce output
+ if (this.persist.get().equals(Persist.NOTHING)) {
+ final Path outputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_G);
+ if (FileSystem.get(this.giraphConfiguration).exists(outputPath)) // TODO: what about when the output is not a file output?
+ FileSystem.get(this.giraphConfiguration).delete(outputPath, true);
+ }
+
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
@@ -230,7 +269,7 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
public Features features() {
return new Features() {
@Override
- public boolean supportsNonSerializableObjects() {
+ public boolean supportsDirectObjects() {
return false;
}
};
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
index 9af0cd0..07cdc51 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
@@ -72,9 +72,9 @@ public class GiraphMemory extends MasterCompute implements Memory {
this.registerPersistentAggregator(key, MemoryAggregator.class);
}
this.registerPersistentAggregator(Constants.GREMLIN_HADOOP_HALT, MemoryAggregator.class);
- this.registerPersistentAggregator(Constants.SYSTEM_RUNTIME, MemoryAggregator.class);
+ this.registerPersistentAggregator(Constants.HIDDEN_RUNTIME, MemoryAggregator.class);
this.setAggregatedValue(Constants.GREMLIN_HADOOP_HALT, new RuleWritable(RuleWritable.Rule.SET, false));
- this.set(Constants.SYSTEM_RUNTIME, System.currentTimeMillis());
+ this.set(Constants.HIDDEN_RUNTIME, System.currentTimeMillis());
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
@@ -103,7 +103,7 @@ public class GiraphMemory extends MasterCompute implements Memory {
@Override
public long getRuntime() {
- return System.currentTimeMillis() - this.<Long>get(Constants.SYSTEM_RUNTIME);
+ return System.currentTimeMillis() - this.<Long>get(Constants.HIDDEN_RUNTIME);
}
@Override
@@ -190,7 +190,7 @@ public class GiraphMemory extends MasterCompute implements Memory {
}
private void checkKeyValue(final String key, final Object value) {
- if (!key.equals(Constants.SYSTEM_RUNTIME) && !this.memoryKeys.contains(key))
+ if (!key.equals(Constants.HIDDEN_RUNTIME) && !this.memoryKeys.contains(key))
throw GraphComputer.Exceptions.providedKeyIsNotAMemoryComputeKey(key);
MemoryHelper.validateValue(value);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/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 db277ae..b578bac 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
@@ -40,11 +40,9 @@ import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
-import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.process.computer.util.DefaultComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
-import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
import org.slf4j.Logger;
@@ -53,6 +51,7 @@ import scala.Tuple2;
import java.io.File;
import java.util.HashSet;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
@@ -71,6 +70,9 @@ public final class SparkGraphComputer implements GraphComputer {
private final Set<MapReduce> mapReducers = new HashSet<>();
private VertexProgram vertexProgram;
+ private Optional<ResultGraph> resultGraph = Optional.empty();
+ private Optional<Persist> persist = Optional.empty();
+
public SparkGraphComputer(final HadoopGraph hadoopGraph) {
this.hadoopGraph = hadoopGraph;
}
@@ -78,7 +80,19 @@ public final class SparkGraphComputer implements GraphComputer {
@Override
public GraphComputer isolation(final Isolation isolation) {
if (!isolation.equals(Isolation.BSP))
- throw GraphComputer.Exceptions.isolationNotSupported(isolation);
+ throw GraphComputer.Exceptions.isolationNotSupported(isolation); // todo: dirty_bsp is when there is no doNothing() call at the end of the round?
+ return this;
+ }
+
+ @Override
+ public GraphComputer result(final ResultGraph resultGraph) {
+ this.resultGraph = Optional.of(resultGraph);
+ return this;
+ }
+
+ @Override
+ public GraphComputer persist(final Persist persist) {
+ this.persist = Optional.of(persist);
return this;
}
@@ -114,8 +128,19 @@ public final class SparkGraphComputer implements GraphComputer {
GraphComputerHelper.validateProgramOnComputer(this, vertexProgram);
this.mapReducers.addAll(this.vertexProgram.getMapReducers());
}
+
+ // determine persistence and result graph options
+ if (!this.persist.isPresent())
+ this.persist = Optional.of(null == this.vertexProgram ? Persist.NOTHING : this.vertexProgram.getPreferredPersist());
+ if (!this.resultGraph.isPresent())
+ this.resultGraph = Optional.of(null == this.vertexProgram ? ResultGraph.ORIGINAL_GRAPH : this.vertexProgram.getPreferredResultGraph());
+ if (this.resultGraph.get().equals(ResultGraph.ORIGINAL_GRAPH))
+ if (!this.persist.get().equals(Persist.NOTHING))
+ throw GraphComputer.Exceptions.resultGraphPersistCombinationNotSupported(this.resultGraph.get(), this.persist.get());
+
// apache and hadoop configurations that are used throughout
final org.apache.commons.configuration.Configuration apacheConfiguration = new HadoopConfiguration(this.hadoopGraph.configuration());
+ apacheConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, this.persist.get().equals(Persist.EDGES));
final Configuration hadoopConfiguration = ConfUtil.makeHadoopConfiguration(apacheConfiguration);
return CompletableFuture.<ComputerResult>supplyAsync(() -> {
@@ -189,12 +214,11 @@ public final class SparkGraphComputer implements GraphComputer {
// write the map reduce output back to disk (memory)
SparkHelper.saveMapReduceRDD(null == reduceRDD ? mapRDD : reduceRDD, mapReduce, finalMemory, hadoopConfiguration);
}
- // close the context or else bad things happen
+ // close the context or else bad things happen // todo: does this happen automatically cause of the try(resource) {} block?
sparkContext.close();
// update runtime and return the newly computed graph
finalMemory.setRuntime(System.currentTimeMillis() - startTime);
- final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram);
- return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : new ComputerGraph(outputGraph, this.vertexProgram.getElementComputeKeys()), finalMemory.asImmutable());
+ return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph, this.resultGraph.get(), this.persist.get()), finalMemory.asImmutable());
}
}
);
@@ -233,7 +257,7 @@ public final class SparkGraphComputer implements GraphComputer {
public Features features() {
return new Features() {
@Override
- public boolean supportsNonSerializableObjects() {
+ public boolean supportsDirectObjects() {
return false;
}
};
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
index 6cd1f49..d8f3f84 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SparkHelper.java
@@ -168,7 +168,7 @@ public final class SparkHelper {
.saveAsNewAPIHadoopFile(outputLocation + "/" + Constants.HIDDEN_G,
NullWritable.class,
VertexWritable.class,
- (Class<OutputFormat<NullWritable, VertexWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class));
+ (Class<OutputFormat<NullWritable, VertexWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class), hadoopConfiguration);
}
}
@@ -179,7 +179,7 @@ public final class SparkHelper {
mapReduceRDD.mapToPair(keyValue -> new Tuple2<>(new ObjectWritable<>(keyValue._1()), new ObjectWritable<>(keyValue._2()))).saveAsNewAPIHadoopFile(outputLocation + "/" + mapReduce.getMemoryKey(),
ObjectWritable.class,
ObjectWritable.class,
- (Class<OutputFormat<ObjectWritable, ObjectWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, OutputFormat.class));
+ (Class<OutputFormat<ObjectWritable, ObjectWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, OutputFormat.class), hadoopConfiguration);
// if its not a SequenceFile there is no certain way to convert to necessary Java objects.
// to get results you have to look through HDFS directory structure. Oh the horror.
try {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
index 8583f44..82fa2df 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
@@ -44,7 +44,7 @@ public final class MemoryMapReduce extends StaticMapReduce<MapReduce.NullObject,
@Override
public String getMemoryKey() {
- return Constants.SYSTEM_MEMORY;
+ return Constants.HIDDEN_MEMORY;
}
private MemoryMapReduce() {
@@ -73,7 +73,7 @@ public final class MemoryMapReduce extends StaticMapReduce<MapReduce.NullObject,
@Override
public void map(final Vertex vertex, final MapEmitter<NullObject, MapMemory> emitter) {
- final MapMemory mapMemory = vertex.<MapMemory>property(Constants.MAP_MEMORY).orElse(new MapMemory());
+ final MapMemory mapMemory = vertex.<MapMemory>property(Constants.GREMLIN_HADOOP_MAP_MEMORY).orElse(new MapMemory());
emitter.emit(mapMemory);
}
@@ -102,7 +102,7 @@ public final class MemoryMapReduce extends StaticMapReduce<MapReduce.NullObject,
@Override
public int hashCode() {
- return (this.getClass().getCanonicalName() + Constants.SYSTEM_MEMORY).hashCode();
+ return (this.getClass().getCanonicalName() + Constants.HIDDEN_MEMORY).hashCode();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONOutputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONOutputFormat.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONOutputFormat.java
index 1acb320..0f3a398 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONOutputFormat.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONOutputFormat.java
@@ -34,11 +34,11 @@ public class GraphSONOutputFormat extends CommonFileOutputFormat {
@Override
public RecordWriter<NullWritable, VertexWritable> getRecordWriter(final TaskAttemptContext job) throws IOException, InterruptedException {
- return new GraphSONRecordWriter(getDataOuputStream(job));
+ return new GraphSONRecordWriter(getDataOuputStream(job), job.getConfiguration());
}
public RecordWriter<NullWritable, VertexWritable> getRecordWriter(final TaskAttemptContext job, final DataOutputStream outputStream) throws IOException, InterruptedException {
- return new GraphSONRecordWriter(outputStream);
+ return new GraphSONRecordWriter(outputStream, job.getConfiguration());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReader.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReader.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReader.java
index f6480c2..ec7649c 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReader.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReader.java
@@ -23,6 +23,7 @@ import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -42,9 +43,10 @@ import java.util.function.Function;
*/
public class GraphSONRecordReader extends RecordReader<NullWritable, VertexWritable> {
- private final GraphSONReader GRAPHSON_READER = GraphSONReader.build().create();
+ private final GraphSONReader graphsonReader = GraphSONReader.build().create();
private final VertexWritable vertexWritable = new VertexWritable();
private final LineRecordReader lineRecordReader;
+ private boolean hasEdges;
public GraphSONRecordReader() {
this.lineRecordReader = new LineRecordReader();
@@ -53,6 +55,7 @@ public class GraphSONRecordReader extends RecordReader<NullWritable, VertexWrita
@Override
public void initialize(final InputSplit genericSplit, final TaskAttemptContext context) throws IOException {
this.lineRecordReader.initialize(genericSplit, context);
+ this.hasEdges = context.getConfiguration().getBoolean(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT_HAS_EDGES, true);
}
@Override
@@ -64,7 +67,9 @@ public class GraphSONRecordReader extends RecordReader<NullWritable, VertexWrita
final Function<DetachedVertex, Vertex> vertexMaker = detachedVertex -> DetachedVertex.addTo(gLocal, detachedVertex);
final Function<DetachedEdge, Edge> edgeMaker = detachedEdge -> DetachedEdge.addTo(gLocal, detachedEdge);
try (InputStream in = new ByteArrayInputStream(this.lineRecordReader.getCurrentValue().getBytes())) {
- this.vertexWritable.set(GRAPHSON_READER.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker));
+ this.vertexWritable.set(this.hasEdges ?
+ this.graphsonReader.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker) :
+ this.graphsonReader.readVertex(in, vertexMaker));
return true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordWriter.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordWriter.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordWriter.java
index d8f50b7..8b3237f 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordWriter.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordWriter.java
@@ -18,12 +18,14 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONWriter;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONWriter;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -35,8 +37,10 @@ import java.io.UnsupportedEncodingException;
public class GraphSONRecordWriter extends RecordWriter<NullWritable, VertexWritable> {
private static final String UTF8 = "UTF-8";
private static final byte[] NEWLINE;
- private final DataOutputStream out;
- private final GraphSONWriter GRAPHSON_WRITER = GraphSONWriter.build().create();
+ private final DataOutputStream outputStream;
+ private final boolean hasEdges;
+ private final GraphSONWriter graphsonWriter = GraphSONWriter.build().create();
+
static {
try {
@@ -46,20 +50,26 @@ public class GraphSONRecordWriter extends RecordWriter<NullWritable, VertexWrita
}
}
- public GraphSONRecordWriter(final DataOutputStream out) {
- this.out = out;
+ public GraphSONRecordWriter(final DataOutputStream outputStream, final Configuration configuration) {
+ this.outputStream = outputStream;
+ this.hasEdges = configuration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true);
}
@Override
public void write(final NullWritable key, final VertexWritable vertex) throws IOException {
if (null != vertex) {
- GRAPHSON_WRITER.writeVertex(out, vertex.get(), Direction.BOTH);
- this.out.write(NEWLINE);
+ if (this.hasEdges) {
+ graphsonWriter.writeVertex(this.outputStream, vertex.get(), Direction.BOTH);
+ this.outputStream.write(NEWLINE);
+ } else {
+ graphsonWriter.writeVertex(this.outputStream, vertex.get());
+ this.outputStream.write(NEWLINE);
+ }
}
}
@Override
public synchronized void close(TaskAttemptContext context) throws IOException {
- this.out.close();
+ this.outputStream.close();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoOutputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoOutputFormat.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoOutputFormat.java
index 1327d90..3b8a26f 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoOutputFormat.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoOutputFormat.java
@@ -34,10 +34,10 @@ public class GryoOutputFormat extends CommonFileOutputFormat {
@Override
public RecordWriter<NullWritable, VertexWritable> getRecordWriter(final TaskAttemptContext job) throws IOException, InterruptedException {
- return new GryoRecordWriter(getDataOuputStream(job));
+ return new GryoRecordWriter(getDataOuputStream(job), job.getConfiguration());
}
public RecordWriter<NullWritable, VertexWritable> getRecordWriter(final TaskAttemptContext job, final DataOutputStream outputStream) throws IOException, InterruptedException {
- return new GryoRecordWriter(outputStream);
+ return new GryoRecordWriter(outputStream, job.getConfiguration());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
index 83e364d..6422104 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
@@ -57,6 +58,7 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
private final GryoReader gryoReader = GryoReader.build().create();
private final VertexWritable vertexWritable = new VertexWritable();
+ private boolean hasEdges;
private long currentLength = 0;
private long splitLength;
@@ -77,6 +79,7 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
this.inputStream = file.getFileSystem(job).open(split.getPath());
this.inputStream.seek(start);
this.splitLength = split.getLength() - (seekToHeader(this.inputStream, start) - start);
+ this.hasEdges = context.getConfiguration().getBoolean(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT_HAS_EDGES, true);
}
private static long seekToHeader(final FSDataInputStream inputStream, final long start) throws IOException {
@@ -131,7 +134,9 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
final Function<DetachedVertex, Vertex> vertexMaker = detachedVertex -> DetachedVertex.addTo(gLocal, detachedVertex);
final Function<DetachedEdge, Edge> edgeMaker = detachedEdge -> DetachedEdge.addTo(gLocal, detachedEdge);
try (InputStream in = new ByteArrayInputStream(output.toByteArray())) {
- this.vertexWritable.set(this.gryoReader.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker));
+ this.vertexWritable.set(this.hasEdges ?
+ this.gryoReader.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker) :
+ this.gryoReader.readVertex(in, vertexMaker));
return true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordWriter.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordWriter.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordWriter.java
index 5e8f8a0..18501d3 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordWriter.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordWriter.java
@@ -18,12 +18,14 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoWriter;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoWriter;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -33,22 +35,27 @@ import java.io.IOException;
*/
public class GryoRecordWriter extends RecordWriter<NullWritable, VertexWritable> {
- private final DataOutputStream out;
- private final GryoWriter GRYO_WRITER = GryoWriter.build().create();
+ private final DataOutputStream outputStream;
+ private final boolean hasEdges;
+ private final GryoWriter gryoWriter = GryoWriter.build().create(); // TODO: configurable build
- public GryoRecordWriter(final DataOutputStream out) {
- this.out = out;
+ public GryoRecordWriter(final DataOutputStream outputStream, final Configuration configuration) {
+ this.outputStream = outputStream;
+ this.hasEdges = configuration.getBoolean(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT_HAS_EDGES, true);
}
@Override
public void write(final NullWritable key, final VertexWritable vertex) throws IOException {
if (null != vertex) {
- GRYO_WRITER.writeVertex(this.out, vertex.get(), Direction.BOTH);
+ if (this.hasEdges)
+ gryoWriter.writeVertex(this.outputStream, vertex.get(), Direction.BOTH);
+ else
+ gryoWriter.writeVertex(this.outputStream, vertex.get());
}
}
@Override
public synchronized void close(final TaskAttemptContext context) throws IOException {
- this.out.close();
+ this.outputStream.close();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/util/HadoopHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/util/HadoopHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/util/HadoopHelper.java
index c6ad68a..e07af1a 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/util/HadoopHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/util/HadoopHelper.java
@@ -22,6 +22,7 @@ import org.apache.commons.configuration.BaseConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.InputOutputHelper;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -31,16 +32,19 @@ public final class HadoopHelper {
private HadoopHelper() {
}
- public static HadoopGraph getOutputGraph(final HadoopGraph hadoopGraph, boolean hasVertexProgram) {
+ public static HadoopGraph getOutputGraph(final HadoopGraph hadoopGraph, final GraphComputer.ResultGraph resultGraph, final GraphComputer.Persist persist) {
final BaseConfiguration newConfiguration = new BaseConfiguration();
newConfiguration.copy(hadoopGraph.configuration());
- if (hasVertexProgram) {
+ if (resultGraph.equals(GraphComputer.ResultGraph.NEW_GRAPH)) {
newConfiguration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, hadoopGraph.configuration().getOutputLocation() + "/" + Constants.HIDDEN_G);
newConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputOutputHelper.getInputFormat(hadoopGraph.configuration().getGraphOutputFormat()).getCanonicalName());
newConfiguration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, hadoopGraph.configuration().getOutputLocation() + "_");
+ newConfiguration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT_HAS_EDGES, persist.equals(GraphComputer.Persist.EDGES));
} else {
newConfiguration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, hadoopGraph.configuration().getOutputLocation() + "_");
}
+
+
return HadoopGraph.open(newConfiguration);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/HadoopGraphProvider.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/HadoopGraphProvider.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/HadoopGraphProvider.java
index e084d3d..50738b2 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/HadoopGraphProvider.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/HadoopGraphProvider.java
@@ -33,6 +33,7 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopProperty;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopVertex;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopVertexProperty;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONInputFormat;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONResourceAccess;
@@ -121,7 +122,7 @@ public class HadoopGraphProvider extends AbstractGraphProvider {
/// spark configuration
put("spark.master", "local[4]");
- put("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+ put("spark.serializer", "org.apache.spark.serializer.JavaSerializer");
}};
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReaderWriterTest.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReaderWriterTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReaderWriterTest.java
index 04fcbf9..4c04bca 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReaderWriterTest.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/graphson/GraphSONRecordReaderWriterTest.java
@@ -28,8 +28,10 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.HadoopGraphProvider;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
@@ -50,9 +52,9 @@ public class GraphSONRecordReaderWriterTest {
@Test
public void testAll() throws Exception {
- Configuration conf = new Configuration(false);
- conf.set("fs.file.impl", LocalFileSystem.class.getName());
- conf.set("fs.default.name", "file:///");
+ Configuration configuration = new Configuration(false);
+ configuration.set("fs.file.impl", LocalFileSystem.class.getName());
+ configuration.set("fs.default.name", "file:///");
File testFile = new File(HadoopGraphProvider.PATHS.get("grateful-dead-vertices.ldjson"));
FileSplit split = new FileSplit(
@@ -60,8 +62,8 @@ public class GraphSONRecordReaderWriterTest {
testFile.length(), null);
System.out.println("reading GraphSON adjacency file " + testFile.getAbsolutePath() + " (" + testFile.length() + " bytes)");
- GraphSONInputFormat inputFormat = ReflectionUtils.newInstance(GraphSONInputFormat.class, conf);
- TaskAttemptContext job = new TaskAttemptContext(conf, new TaskAttemptID());
+ GraphSONInputFormat inputFormat = ReflectionUtils.newInstance(GraphSONInputFormat.class, configuration);
+ TaskAttemptContext job = new TaskAttemptContext(configuration, new TaskAttemptID());
RecordReader reader = inputFormat.createRecordReader(split, job);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReaderWriterTest.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReaderWriterTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReaderWriterTest.java
index e1a661a..c0c98ef 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReaderWriterTest.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReaderWriterTest.java
@@ -49,9 +49,9 @@ import static org.junit.Assert.assertTrue;
public class GryoRecordReaderWriterTest {
@Test
public void testAll() throws Exception {
- Configuration conf = new Configuration(false);
- conf.set("fs.file.impl", LocalFileSystem.class.getName());
- conf.set("fs.default.name", "file:///");
+ Configuration configuration = new Configuration(false);
+ configuration.set("fs.file.impl", LocalFileSystem.class.getName());
+ configuration.set("fs.default.name", "file:///");
File testFile = new File(HadoopGraphProvider.PATHS.get("grateful-dead-vertices.kryo"));
FileSplit split = new FileSplit(
@@ -59,8 +59,8 @@ public class GryoRecordReaderWriterTest {
testFile.length(), null);
System.out.println("reading Gryo file " + testFile.getAbsolutePath() + " (" + testFile.length() + " bytes)");
- GryoInputFormat inputFormat = ReflectionUtils.newInstance(GryoInputFormat.class, conf);
- TaskAttemptContext job = new TaskAttemptContext(conf, new TaskAttemptID());
+ GryoInputFormat inputFormat = ReflectionUtils.newInstance(GryoInputFormat.class, configuration);
+ TaskAttemptContext job = new TaskAttemptContext(configuration, new TaskAttemptID());
RecordReader reader = inputFormat.createRecordReader(split, job);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a0d2f08..cef9e9b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -243,6 +243,7 @@ limitations under the License.
<exclude>**/src/main/resources/org/apache/tinkerpop/gremlin/structure/io/script/tinkerpop-classic.txt</exclude>
<exclude>**/src/main/resources/META-INF/services/**</exclude>
<exclude>**/src/main/ext/**</exclude>
+ <exclude>**/_bsp/**</exclude>
</excludes>
<licenses>
<license implementation="org.apache.rat.analysis.license.ApacheSoftwareLicense20"/>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/5a4428f7/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
----------------------------------------------------------------------
diff --git a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
index 779fa41..afd36b2 100644
--- a/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
+++ b/tinkergraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/tinkergraph/process/computer/TinkerGraphComputer.java
@@ -34,6 +34,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -46,6 +47,9 @@ import java.util.concurrent.Future;
public class TinkerGraphComputer implements GraphComputer {
private Isolation isolation = Isolation.BSP;
+ private Optional<ResultGraph> resultGraph = Optional.empty();
+ private Optional<Persist> persist = Optional.empty();
+
private VertexProgram<?> vertexProgram;
private final TinkerGraph graph;
private TinkerMemory memory;
@@ -53,6 +57,7 @@ public class TinkerGraphComputer implements GraphComputer {
private boolean executed = false;
private final Set<MapReduce> mapReducers = new HashSet<>();
+
public TinkerGraphComputer(final TinkerGraph graph) {
this.graph = graph;
}
@@ -68,6 +73,18 @@ public class TinkerGraphComputer implements GraphComputer {
}
@Override
+ public GraphComputer result(final ResultGraph resultGraph) {
+ this.resultGraph = Optional.of(resultGraph);
+ return this;
+ }
+
+ @Override
+ public GraphComputer persist(final Persist persist) {
+ this.persist = Optional.of(persist);
+ return this;
+ }
+
+ @Override
public GraphComputer program(final VertexProgram vertexProgram) {
this.vertexProgram = vertexProgram;
return this;
@@ -94,6 +111,15 @@ public class TinkerGraphComputer implements GraphComputer {
GraphComputerHelper.validateProgramOnComputer(this, this.vertexProgram);
this.mapReducers.addAll(this.vertexProgram.getMapReducers());
}
+
+ if (!this.persist.isPresent())
+ this.persist = Optional.of(null == this.vertexProgram ? Persist.NOTHING : this.vertexProgram.getPreferredPersist());
+ if (!this.resultGraph.isPresent())
+ this.resultGraph = Optional.of(null == this.vertexProgram ? ResultGraph.ORIGINAL_GRAPH : this.vertexProgram.getPreferredResultGraph());
+ if (this.resultGraph.get().equals(ResultGraph.ORIGINAL_GRAPH))
+ if (!this.persist.get().equals(Persist.NOTHING))
+ throw GraphComputer.Exceptions.resultGraphPersistCombinationNotSupported(this.resultGraph.get(), this.persist.get());
+
//final Graph computeGraph = this.graph;
final ComputerGraph computeGraph = new ComputerGraph(this.graph, null == this.vertexProgram ? Collections.emptySet() : this.vertexProgram.getElementComputeKeys());
this.memory = new TinkerMemory(this.vertexProgram, this.mapReducers);
@@ -174,8 +200,7 @@ public class TinkerGraphComputer implements GraphComputer {
// update runtime and return the newly computed graph
this.memory.setRuntime(System.currentTimeMillis() - time);
this.memory.complete();
- computeGraph.setState(ComputerGraph.State.NO_OP);
- return new TinkerComputerResult(computeGraph, this.memory.asImmutable());
+ return new TinkerComputerResult(this.graph, this.memory.asImmutable());
} catch (Exception ex) {
throw new RuntimeException(ex);