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/05/27 19:58:06 UTC
[2/2] incubator-tinkerpop git commit: GiraphGraphComputer no longer
requires another BSP round to compute its memory. This removed alot of extra
code, extra MapReduce,
and numerous Constants. Added more tests to GraphComputerTest that verify
Persist and
GiraphGraphComputer no longer requires another BSP round to compute its memory. This removed alot of extra code, extra MapReduce, and numerous Constants. Added more tests to GraphComputerTest that verify Persist and ResultGraph options.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/ec9365a2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/ec9365a2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/ec9365a2
Branch: refs/heads/master
Commit: ec9365a21f38fc44f791f40c6949328c55a6f0ca
Parents: 8a027df
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed May 27 11:58:13 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed May 27 11:58:13 2015 -0600
----------------------------------------------------------------------
docs/src/implementations.asciidoc | 1 -
.../ranking/PageRankVertexProgramTest.java | 55 +++++----
hadoop-gremlin/conf/hadoop-graphson.properties | 2 -
hadoop-gremlin/conf/hadoop-gryo.properties | 1 -
hadoop-gremlin/conf/hadoop-script.properties | 1 -
.../tinkerpop/gremlin/hadoop/Constants.java | 7 --
.../computer/giraph/GiraphComputeVertex.java | 16 +--
.../computer/giraph/GiraphGraphComputer.java | 32 ++---
.../process/computer/giraph/GiraphMemory.java | 58 ++++++---
.../computer/giraph/GiraphWorkerContext.java | 7 --
.../process/computer/util/MemoryMapReduce.java | 119 -------------------
.../gremlin/hadoop/HadoopGraphProvider.java | 2 +-
.../process/computer/spark/io/InputRDDTest.java | 2 +-
13 files changed, 92 insertions(+), 211 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/docs/src/implementations.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/implementations.asciidoc b/docs/src/implementations.asciidoc
index 990a170..5201f00 100644
--- a/docs/src/implementations.asciidoc
+++ b/docs/src/implementations.asciidoc
@@ -571,7 +571,6 @@ gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
gremlin.hadoop.graphInputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat
gremlin.hadoop.graphOutputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat
gremlin.hadoop.memoryOutputFormat=org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
-gremlin.hadoop.deriveMemory=false
gremlin.hadoop.jarsInDistributedCache=true
gremlin.hadoop.inputLocation=tinkerpop-modern-vertices.kryo
gremlin.hadoop.outputLocation=output
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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 d4611f1..df65c7b 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
@@ -20,13 +20,14 @@ package org.apache.tinkerpop.gremlin.process.computer.ranking;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank.PageRankVertexProgram;
import org.junit.Test;
import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -36,30 +37,32 @@ public class PageRankVertexProgramTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void shouldExecutePageRank() throws Exception {
- final ComputerResult result = graph.compute(g.getGraphComputer().get().getClass()).program(PageRankVertexProgram.build().create(graph)).submit().get();
- result.graph().traversal().V().forEachRemaining(v -> {
- assertTrue(v.keys().contains("name"));
- 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);
- if (name.equals("marko"))
- assertTrue(pageRank > 0.14 && pageRank < 0.16);
- else if (name.equals("vadas"))
- assertTrue(pageRank > 0.19 && pageRank < 0.20);
- else if (name.equals("lop"))
- assertTrue(pageRank > 0.40 && pageRank < 0.41);
- else if (name.equals("josh"))
- assertTrue(pageRank > 0.19 && pageRank < 0.20);
- else if (name.equals("ripple"))
- assertTrue(pageRank > 0.23 && pageRank < 0.24);
- else if (name.equals("peter"))
- assertTrue(pageRank > 0.14 && pageRank < 0.16);
- else
- throw new IllegalStateException("The following vertex should not exist in the graph: " + name);
- });
- assertEquals(result.memory().getIteration(), 30);
- assertEquals(result.memory().asMap().size(), 0);
+ if (g.getGraphComputer().get().features().supportsResultGraphPersistCombination(GraphComputer.ResultGraph.NEW, GraphComputer.Persist.VERTEX_PROPERTIES)) {
+ final ComputerResult result = graph.compute(g.getGraphComputer().get().getClass()).program(PageRankVertexProgram.build().create(graph)).submit().get();
+ result.graph().traversal().V().forEachRemaining(v -> {
+ assertTrue(v.keys().contains("name"));
+ 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);
+ if (name.equals("marko"))
+ assertTrue(pageRank > 0.14 && pageRank < 0.16);
+ else if (name.equals("vadas"))
+ assertTrue(pageRank > 0.19 && pageRank < 0.20);
+ else if (name.equals("lop"))
+ assertTrue(pageRank > 0.40 && pageRank < 0.41);
+ else if (name.equals("josh"))
+ assertTrue(pageRank > 0.19 && pageRank < 0.20);
+ else if (name.equals("ripple"))
+ assertTrue(pageRank > 0.23 && pageRank < 0.24);
+ else if (name.equals("peter"))
+ assertTrue(pageRank > 0.14 && pageRank < 0.16);
+ else
+ throw new IllegalStateException("The following vertex should not exist in the graph: " + name);
+ });
+ assertEquals(result.memory().getIteration(), 30);
+ assertEquals(result.memory().asMap().size(), 0);
+ }
}
/*@Test
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/hadoop-gremlin/conf/hadoop-graphson.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-graphson.properties b/hadoop-gremlin/conf/hadoop-graphson.properties
index 146a015..2013b7b 100644
--- a/hadoop-gremlin/conf/hadoop-graphson.properties
+++ b/hadoop-gremlin/conf/hadoop-graphson.properties
@@ -24,8 +24,6 @@ gremlin.hadoop.memoryOutputFormat=org.apache.hadoop.mapreduce.lib.output.TextOut
# i/o locations
gremlin.hadoop.inputLocation=tinkerpop-modern.json
gremlin.hadoop.outputLocation=output
-# deriving a complete view of the memory requires an extra mapreduce job and thus, if not needed, should be avoided
-gremlin.hadoop.deriveMemory=false
# if the job jars are not on the classpath of every hadoop node, then they must be provided to the distributed cache at runtime
gremlin.hadoop.jarsInDistributedCache=true
# the vertex program to execute
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/hadoop-gremlin/conf/hadoop-gryo.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-gryo.properties b/hadoop-gremlin/conf/hadoop-gryo.properties
index 13a1e41..2eba291 100644
--- a/hadoop-gremlin/conf/hadoop-gryo.properties
+++ b/hadoop-gremlin/conf/hadoop-gryo.properties
@@ -18,7 +18,6 @@ gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
gremlin.hadoop.graphInputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoInputFormat
gremlin.hadoop.graphOutputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat
gremlin.hadoop.memoryOutputFormat=org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
-gremlin.hadoop.deriveMemory=false
gremlin.hadoop.jarsInDistributedCache=true
gremlin.hadoop.inputLocation=tinkerpop-modern.kryo
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/hadoop-gremlin/conf/hadoop-script.properties
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/conf/hadoop-script.properties b/hadoop-gremlin/conf/hadoop-script.properties
index 0388801..0792023 100644
--- a/hadoop-gremlin/conf/hadoop-script.properties
+++ b/hadoop-gremlin/conf/hadoop-script.properties
@@ -18,7 +18,6 @@ gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
gremlin.hadoop.graphInputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat
gremlin.hadoop.graphOutputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONOutputFormat
gremlin.hadoop.memoryOutputFormat=org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
-gremlin.hadoop.deriveMemory=false
gremlin.hadoop.jarsInDistributedCache=true
gremlin.hadoop.inputLocation=tinkerpop-classic.txt
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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 b2e1b0a..de744b7 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
@@ -36,7 +36,6 @@ public final class Constants {
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_JARS_IN_DISTRIBUTED_CACHE = "gremlin.hadoop.jarsInDistributedCache";
public static final String HIDDEN_G = Graph.Hidden.hide("g");
@@ -46,14 +45,8 @@ public final class Constants {
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 HIDDEN_MEMORY = Graph.Hidden.hide("memory");
- 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 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/ec9365a2/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 1507def..2581388 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
@@ -20,14 +20,10 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
import org.apache.giraph.graph.Vertex;
import org.apache.hadoop.io.NullWritable;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.hadoop.process.computer.util.Rule;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
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.MapMemory;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -48,17 +44,7 @@ public final class GiraphComputeVertex extends Vertex<ObjectWritable, VertexWrit
public void compute(final Iterable<ObjectWritable> messages) {
final GiraphWorkerContext workerContext = (GiraphWorkerContext) this.getWorkerContext();
final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
- final GiraphMemory memory = workerContext.getMemory();
- final GiraphMessenger messenger = workerContext.getMessenger(this, messages.iterator());
- ///////////
- if (!(Boolean) ((Rule) this.getAggregatedValue(Constants.GREMLIN_HADOOP_HALT)).getObject()) {
- vertexProgram.execute(ComputerGraph.vertexProgram(this.getValue().get(), vertexProgram), messenger, memory);
- } else if (workerContext.deriveMemory()) {
- final MapMemory mapMemory = new MapMemory();
- memory.asMap().forEach(mapMemory::set);
- mapMemory.setIteration(memory.getIteration() - 1);
- this.getValue().get().property(VertexProperty.Cardinality.single, Constants.GREMLIN_HADOOP_MAP_MEMORY, mapMemory); // not a compute key cause no ComputerGraph is used
- }
+ vertexProgram.execute(ComputerGraph.vertexProgram(this.getValue().get(), vertexProgram), workerContext.getMessenger(this, messages.iterator()), workerContext.getMemory());
workerContext.getVertexProgramPool().offer(vertexProgram);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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 e51b741..e8f026f 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,7 +31,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputFormat;
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;
@@ -39,9 +38,9 @@ import org.apache.tinkerpop.gremlin.hadoop.process.computer.AbstractHadoopGraphC
import org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph.io.GiraphVertexInputFormat;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph.io.GiraphVertexOutputFormat;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.util.MapReduceHelper;
-import org.apache.tinkerpop.gremlin.hadoop.process.computer.util.MemoryMapReduce;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritableIterator;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.HadoopHelper;
@@ -54,8 +53,6 @@ import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import java.io.File;
import java.io.NotSerializableException;
-import java.util.HashSet;
-import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.stream.Stream;
@@ -108,6 +105,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
//e.printStackTrace();
throw new IllegalStateException(e.getMessage(), e);
}
+
this.memory.setRuntime(System.currentTimeMillis() - startTime);
return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph, this.resultGraph.get(), this.persist.get()), this.memory.asImmutable());
});
@@ -126,14 +124,6 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
if (e.getCause() instanceof NumberFormatException)
throw new NotSerializableException("The provided traversal is not serializable and thus, can not be distributed across the cluster");
}
- // 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<>(this.vertexProgram.getMemoryComputeKeys());
- memoryKeys.add(Constants.HIDDEN_ITERATION);
- this.giraphConfiguration.setStrings(Constants.GREMLIN_HADOOP_MEMORY_KEYS, (String[]) memoryKeys.toArray(new String[memoryKeys.size()]));
- this.mapReducers.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);
// handle input paths (if any)
@@ -145,15 +135,24 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
}
// handle output paths
final Path outputPath = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_G);
- if (this.persist.get().equals(Persist.NOTHING) && this.mapReducers.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);
+ FileOutputFormat.setOutputPath(job.getInternalJob(), outputPath);
job.getInternalJob().setJarByClass(GiraphGraphComputer.class);
this.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"); // how do I get the exception that occured?
+ // add vertex program memory values to the return memory
+ for (final String key : this.vertexProgram.getMemoryComputeKeys()) {
+ final Path path = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + key);
+ final ObjectWritableIterator iterator = new ObjectWritableIterator(this.giraphConfiguration, path);
+ if (iterator.hasNext()) {
+ this.memory.set(key, iterator.next().getValue());
+ }
+ FileSystem.get(this.giraphConfiguration).delete(path, true);
+ }
+ final Path path = new Path(this.giraphConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_ITERATION);
+ this.memory.setIteration((Integer) new ObjectWritableIterator(this.giraphConfiguration, path).next().getValue());
+ FileSystem.get(this.giraphConfiguration).delete(path, true);
}
// 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));
@@ -161,6 +160,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
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);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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 78e0e45..2025914 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
@@ -20,21 +20,29 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
import org.apache.commons.configuration.Configuration;
import org.apache.giraph.master.MasterCompute;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.util.Rule;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
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.MapMemory;
import org.apache.tinkerpop.gremlin.process.computer.util.MemoryHelper;
-import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.io.DataInput;
import java.io.DataOutput;
import java.util.HashSet;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -76,23 +84,39 @@ public final class GiraphMemory extends MasterCompute implements Memory {
MemoryHelper.validateKey(key);
this.registerPersistentAggregator(key, MemoryAggregator.class);
}
- this.registerPersistentAggregator(Constants.GREMLIN_HADOOP_HALT, MemoryAggregator.class);
- this.registerPersistentAggregator(Constants.HIDDEN_RUNTIME, MemoryAggregator.class);
- this.setAggregatedValue(Constants.GREMLIN_HADOOP_HALT, new Rule(Rule.Operation.SET, Boolean.FALSE));
- this.set(Constants.HIDDEN_RUNTIME, System.currentTimeMillis());
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
this.vertexProgram.setup(this);
} else {
- if (this.get(Constants.GREMLIN_HADOOP_HALT)) {
+ if (this.vertexProgram.terminate(this)) { // terminate
+ // write the memory to HDFS
+ final MapMemory memory = new MapMemory(this);
+ // a hack to get the last iteration memory values to stick
+ this.vertexProgram.terminate(memory);
+ final String outputLocation = this.getConf().get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, null);
+ if (null != outputLocation) {
+ try {
+ final Class<? extends OutputFormat> memoryOutputFormat = this.getConf().getClass(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, SequenceFileOutputFormat.class, OutputFormat.class);
+ if (!memoryOutputFormat.equals(SequenceFileOutputFormat.class))
+ HadoopGraph.LOGGER.warn(Constants.SEQUENCE_WARNING);
+ else {
+ for (final String key : this.keys()) {
+ final SequenceFile.Writer writer = SequenceFile.createWriter(FileSystem.get(this.getConf()), this.getConf(), new Path(outputLocation + "/" + key), ObjectWritable.class, ObjectWritable.class);
+ writer.append(new ObjectWritable<>(MapReduce.NullObject.instance()), new ObjectWritable<>(memory.get(key)));
+ writer.close();
+ }
+ final SequenceFile.Writer writer = SequenceFile.createWriter(FileSystem.get(this.getConf()), this.getConf(), new Path(outputLocation + "/" + Constants.HIDDEN_ITERATION), ObjectWritable.class, ObjectWritable.class);
+ writer.append(new ObjectWritable<>(MapReduce.NullObject.instance()), new ObjectWritable<>(memory.getIteration()));
+ writer.close();
+ }
+ } catch (final Exception e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
this.haltComputation();
- } else if (this.vertexProgram.terminate(this)) { // terminate
- if (!this.getConf().getBoolean(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, false)) // no need for the extra BSP round if memory is not required
- this.haltComputation();
- else
- this.setAggregatedValue(Constants.GREMLIN_HADOOP_HALT, new Rule(Rule.Operation.SET, Boolean.TRUE));
}
+
}
}
@@ -108,12 +132,18 @@ public final class GiraphMemory extends MasterCompute implements Memory {
@Override
public long getRuntime() {
- return System.currentTimeMillis() - this.<Long>get(Constants.HIDDEN_RUNTIME);
+ return System.currentTimeMillis(); // TODO: this should be stored in the configuration
}
@Override
public Set<String> keys() {
- return this.memoryKeys;
+ final Set<String> keys = new HashSet<>();
+ for(final String key : this.memoryKeys) {
+ if(this.exists(key))
+ keys.add(key);
+ }
+ return keys;
+ // return this.memoryKeys.stream().filter(this::exists).collect(Collectors.toSet());
}
@Override
@@ -195,7 +225,7 @@ public final class GiraphMemory extends MasterCompute implements Memory {
}
private void checkKeyValue(final String key, final Object value) {
- if (!key.equals(Constants.HIDDEN_RUNTIME) && !this.memoryKeys.contains(key))
+ if (!this.memoryKeys.contains(key))
throw GraphComputer.Exceptions.providedKeyIsNotAMemoryComputeKey(key);
MemoryHelper.validateValue(value);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
index cd42fa0..85d2c2d 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
@@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
import org.apache.commons.configuration.Configuration;
import org.apache.giraph.conf.GiraphConstants;
import org.apache.giraph.worker.WorkerContext;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.HadoopPools;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
@@ -39,7 +38,6 @@ public final class GiraphWorkerContext extends WorkerContext {
private VertexProgramPool vertexProgramPool;
private GiraphMemory memory;
- private boolean deriveMemory;
public GiraphWorkerContext() {
// Giraph ReflectionUtils requires this to be public at minimum
@@ -51,7 +49,6 @@ public final class GiraphWorkerContext extends WorkerContext {
final VertexProgram vertexProgram = VertexProgram.createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.vertexProgramPool = new VertexProgramPool(vertexProgram, this.getContext().getConfiguration().getInt(GiraphConstants.NUM_COMPUTE_THREADS.getKey(), 1));
this.memory = new GiraphMemory(this, vertexProgram);
- this.deriveMemory = this.getContext().getConfiguration().getBoolean(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, false);
}
public void postApplication() {
@@ -77,8 +74,4 @@ public final class GiraphWorkerContext extends WorkerContext {
public GiraphMessenger getMessenger(final GiraphComputeVertex giraphComputeVertex, final Iterator<ObjectWritable> messages) {
return new GiraphMessenger(giraphComputeVertex, messages);
}
-
- public boolean deriveMemory() {
- return this.deriveMemory;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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
deleted file mode 100644
index 3afdc19..0000000
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.tinkerpop.gremlin.hadoop.process.computer.util;
-
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
-import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
-import org.apache.tinkerpop.gremlin.process.computer.Memory;
-import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
-import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
-import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.commons.configuration.Configuration;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class MemoryMapReduce extends StaticMapReduce<MapReduce.NullObject, MapMemory, MapReduce.NullObject, MapMemory, MapMemory> {
-
- public Set<String> memoryKeys = new HashSet<>();
-
- @Override
- public String getMemoryKey() {
- return Constants.HIDDEN_MEMORY;
- }
-
- private MemoryMapReduce() {
-
- }
-
- public MemoryMapReduce(final Set<String> memoryKeys) {
- this.memoryKeys = memoryKeys;
- }
-
- @Override
- public void storeState(final Configuration configuration) {
- super.storeState(configuration);
- configuration.setProperty(Constants.GREMLIN_HADOOP_MEMORY_KEYS, new ArrayList<>(this.memoryKeys.size()));
- }
-
- @Override
- public void loadState(final Graph graph, final Configuration configuration) {
- this.memoryKeys = new HashSet((List) configuration.getList(Constants.GREMLIN_HADOOP_MEMORY_KEYS));
- }
-
- @Override
- public boolean doStage(final Stage stage) {
- return true;
- }
-
- @Override
- public void map(final Vertex vertex, final MapEmitter<NullObject, MapMemory> emitter) {
- final MapMemory mapMemory = vertex.<MapMemory>property(Constants.GREMLIN_HADOOP_MAP_MEMORY).orElse(new MapMemory());
- emitter.emit(mapMemory);
- }
-
- @Override
- public void combine(final NullObject key, final Iterator<MapMemory> values, final ReduceEmitter<NullObject, MapMemory> emitter) {
- this.reduce(key, values, emitter);
- }
-
- @Override
- public void reduce(final NullObject key, final Iterator<MapMemory> values, final ReduceEmitter<NullObject, MapMemory> emitter) {
- emitter.emit(key, values.next());
- }
-
- @Override
- public MapMemory generateFinalResult(final Iterator<KeyValue<NullObject, MapMemory>> keyValues) {
- return keyValues.next().getValue();
- }
-
- @Override
- public void addResultToMemory(final Memory.Admin memory, final Iterator<KeyValue<NullObject, MapMemory>> keyValues) {
- final MapMemory temp = keyValues.next().getValue();
- temp.asMap().forEach(memory::set);
- memory.setIteration(temp.getIteration());
- memory.setRuntime(temp.getRuntime());
- }
-
- @Override
- public int hashCode() {
- return (this.getClass().getCanonicalName() + Constants.HIDDEN_MEMORY).hashCode();
- }
-
- @Override
- public boolean equals(final Object object) {
- return GraphComputerHelper.areEqual(this, object);
- }
-
- @Override
- public String toString() {
- return StringFactory.mapReduceString(this, this.memoryKeys.toString());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/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 b593c8a..0fab71d 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
@@ -111,7 +111,7 @@ public class HadoopGraphProvider extends AbstractGraphProvider {
put(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, GryoOutputFormat.class.getCanonicalName());
put(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, SequenceFileOutputFormat.class.getCanonicalName());
put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
- put(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, true);
+ // put(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, true);
put(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
/// giraph configuration
put(GiraphConstants.MIN_WORKERS, 1);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/ec9365a2/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/io/InputRDDTest.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/io/InputRDDTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/io/InputRDDTest.java
index d245a95..09df899 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/io/InputRDDTest.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/io/InputRDDTest.java
@@ -50,7 +50,7 @@ public class InputRDDTest {
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, GryoOutputFormat.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, SequenceFileOutputFormat.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
- configuration.setProperty(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, true);
+ // configuration.setProperty(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, true);
configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
////////
Graph graph = GraphFactory.open(configuration);