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/17 17:07:54 UTC
incubator-tinkerpop git commit: Fixed a bug in InputOutput conversion
with the Hadoop XXXGraphComputers. Greatly reduce the code complexity of
GryoInputFormat.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master ac0738e09 -> 8798d8280
Fixed a bug in InputOutput conversion with the Hadoop XXXGraphComputers. Greatly reduce the code complexity of GryoInputFormat.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/8798d828
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/8798d828
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/8798d828
Branch: refs/heads/master
Commit: 8798d828081180feba8821dbf849bfc02c35bf00
Parents: ac0738e
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Mar 17 10:07:49 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Mar 17 10:07:49 2015 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 2 +-
.../traversal/step/filter/GroovyHasTest.groovy | 4 +-
.../graph/traversal/step/filter/HasTest.java | 8 +-
.../tinkerpop/gremlin/hadoop/Constants.java | 2 +-
.../computer/example/TraversalSupplier1.java | 2 +-
.../computer/example/TraversalSupplier2.java | 2 +-
.../computer/example/TraversalSupplier3.java | 2 +-
.../computer/giraph/GiraphGraphComputer.java | 4 +-
.../computer/spark/SparkGraphComputer.java | 10 +-
.../spark/util/SApacheConfiguration.java | 70 ----------
.../computer/spark/util/SparkHelper.java | 2 +-
.../process/computer/util/MapReduceHelper.java | 2 +-
.../hadoop/structure/HadoopConfiguration.java | 73 +++++-----
.../hadoop/structure/io/InputOutputHelper.java | 44 +++---
.../io/graphson/GraphSONRecordReader.java | 14 +-
.../structure/io/gryo/GryoRecordReader.java | 66 +++++++--
.../structure/io/gryo/GryoRecordWriter.java | 2 +-
.../structure/io/gryo/VertexStreamIterator.java | 138 -------------------
.../hadoop/structure/util/HadoopHelper.java | 13 +-
.../io/gryo/GryoRecordReaderWriterTest.java | 2 +-
.../io/gryo/VertexStreamIteratorTest.java | 99 -------------
21 files changed, 158 insertions(+), 403 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index e42f4be..0a7e21f 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -57,7 +57,7 @@ TinkerPop 3.0.0.M8 (Release Date: NOT OFFICIALLY RELEASED YET)
* Bumped `neo4j-graph` to Neo4j 2.1.6.
* Bumped to Groovy 2.4.1.
* Added a new "performance" test suite for Gremlin Process.
-* Steps now only operate with traversals -- no more lambdas. Lambda->`Traversal` conversion utilities added.
+* Steps now only operate with traversals -- no more lambdas. Lambda->`Traversal` conversion utilities added.
* `SideEffectStep` always requires a `Consumer`. Steps that were consumer-less simply extends `AbstractStep`.
* Simplified the `Neo4jGraph` implementation by now allowing `cypher()` mid-traversal. Only available via `g.cypher()`.
* Moved `clock()` out of the Utility plugin. It is now available to both Groovy and Java.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/GroovyHasTest.groovy
----------------------------------------------------------------------
diff --git a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/GroovyHasTest.groovy b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/GroovyHasTest.groovy
index 9ce7975..a591901 100644
--- a/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/GroovyHasTest.groovy
+++ b/gremlin-groovy-test/src/main/groovy/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/GroovyHasTest.groovy
@@ -35,7 +35,7 @@ public abstract class GroovyHasTest {
public static class StandardTraversals extends HasTest {
@Override
- public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX() {
+ public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_name() {
g.V.out('knows').has(__.out('created')).name
}
@@ -114,7 +114,7 @@ public abstract class GroovyHasTest {
public static class ComputerTraversals extends HasTest {
@Override
- public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX() {
+ public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_name() {
ComputerTestHelper.compute("g.V.out('knows').has(__.out('created')).name", g);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/HasTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/HasTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/HasTest.java
index 5caf72a..f7165da 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/HasTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/graph/traversal/step/filter/HasTest.java
@@ -20,12 +20,10 @@ package org.apache.tinkerpop.gremlin.process.graph.traversal.step.filter;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
-import org.apache.tinkerpop.gremlin.process.T;
import org.apache.tinkerpop.gremlin.process.Traversal;
import org.apache.tinkerpop.gremlin.process.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.UseEngine;
import org.apache.tinkerpop.gremlin.structure.Compare;
-import org.apache.tinkerpop.gremlin.structure.Contains;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.Vertex;
@@ -48,7 +46,7 @@ import static org.junit.Assume.assumeTrue;
*/
public abstract class HasTest extends AbstractGremlinProcessTest {
- public abstract Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX();
+ public abstract Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_name();
public abstract Traversal<Vertex, Vertex> get_g_VX1X_hasXkeyX(final Object v1Id, final String key);
@@ -81,7 +79,7 @@ public abstract class HasTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX() {
- Traversal<Vertex, String> traversal = get_g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX();
+ Traversal<Vertex, String> traversal = get_g_V_outXknowsX_hasXoutXcreatedXX_name();
printTraversalForm(traversal);
checkResults(Arrays.asList("josh"), traversal);
}
@@ -250,7 +248,7 @@ public abstract class HasTest extends AbstractGremlinProcessTest {
@UseEngine(TraversalEngine.Type.COMPUTER)
public static class Traversals extends HasTest {
@Override
- public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_valuesXnameX() {
+ public Traversal<Vertex, String> get_g_V_outXknowsX_hasXoutXcreatedXX_name() {
return g.V().out("knows").has(out("created")).values("name");
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 697cab0..5062bb4 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
@@ -33,7 +33,7 @@ public class Constants {
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 SYSTEM_G = Graph.Hidden.hide("g");
+ 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): ";
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier1.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier1.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier1.java
index edca01f..13ebd04 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier1.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier1.java
@@ -29,6 +29,6 @@ import java.util.function.Supplier;
public class TraversalSupplier1 implements Supplier<Traversal> {
@Override
public Traversal get() {
- return HadoopGraph.open().traversal().V().out().out().values("name");
+ return HadoopGraph.empty().traversal().V().out().out().values("name");
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier2.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier2.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier2.java
index d3477f8..6714edd 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier2.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier2.java
@@ -29,6 +29,6 @@ import java.util.function.Supplier;
public class TraversalSupplier2 implements Supplier<Traversal> {
@Override
public Traversal get() {
- return HadoopGraph.open().traversal().V().<String>values("name").map(s -> s.get().length()).groupCount().<Integer>by(i -> i + 100);
+ return HadoopGraph.empty().traversal().V().<String>values("name").map(s -> s.get().length()).groupCount().<Integer>by(i -> i + 100);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier3.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier3.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier3.java
index efec301..5089bf5 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier3.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/example/TraversalSupplier3.java
@@ -30,6 +30,6 @@ import java.util.function.Supplier;
public class TraversalSupplier3 implements Supplier<Traversal> {
@Override
public Traversal get() {
- return HadoopGraph.open().traversal().V().<String>values("name").group().<String>by(s -> s.substring(1, 2)).by(v -> v).<Collection>by(Collection::size);
+ return HadoopGraph.empty().traversal().V().<String>values("name").group().<String>by(s -> s.substring(1, 2)).by(v -> v).<Collection>by(Collection::size);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 4c4558a..1320c8f 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
@@ -148,7 +148,7 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
}
this.memory.setRuntime(System.currentTimeMillis() - startTime);
- final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph);
+ final Graph outputGraph = HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram);
return new DefaultComputerResult(null == this.vertexProgram ? outputGraph : ComputerDataStrategy.wrapGraph(outputGraph, this.vertexProgram), this.memory.asImmutable());
});
}
@@ -163,7 +163,7 @@ public class GiraphGraphComputer extends Configured implements GraphComputer, To
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.SYSTEM_G));
+ 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)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 a8ad2f4..cd243e5 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
@@ -29,8 +29,8 @@ import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.util.SApacheConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.util.SparkHelper;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
@@ -113,7 +113,7 @@ public final class SparkGraphComputer implements GraphComputer {
this.mapReducers.addAll(this.vertexProgram.getMapReducers());
}
// apache and hadoop configurations that are used throughout
- final org.apache.commons.configuration.Configuration apacheConfiguration = new SApacheConfiguration(this.hadoopGraph.configuration());
+ final org.apache.commons.configuration.Configuration apacheConfiguration = new HadoopConfiguration(this.hadoopGraph.configuration());
final Configuration hadoopConfiguration = ConfUtil.makeHadoopConfiguration(apacheConfiguration);
return CompletableFuture.<ComputerResult>supplyAsync(() -> {
@@ -147,7 +147,7 @@ public final class SparkGraphComputer implements GraphComputer {
memory = new SparkMemory(this.vertexProgram, this.mapReducers, sparkContext);
this.vertexProgram.setup(memory);
memory.broadcastMemory(sparkContext);
- final SApacheConfiguration vertexProgramConfiguration = new SApacheConfiguration();
+ final HadoopConfiguration vertexProgramConfiguration = new HadoopConfiguration();
this.vertexProgram.storeState(vertexProgramConfiguration);
ConfigurationUtils.copy(vertexProgramConfiguration, apacheConfiguration);
ConfUtil.mergeApacheIntoHadoopConfiguration(vertexProgramConfiguration, hadoopConfiguration);
@@ -177,7 +177,7 @@ public final class SparkGraphComputer implements GraphComputer {
final Memory.Admin finalMemory = null == memory ? new MapMemory() : new MapMemory(memory);
for (final MapReduce mapReduce : this.mapReducers) {
// execute the map reduce job
- final SApacheConfiguration newApacheConfiguration = new SApacheConfiguration(apacheConfiguration);
+ final HadoopConfiguration newApacheConfiguration = new HadoopConfiguration(apacheConfiguration);
mapReduce.storeState(newApacheConfiguration);
// map
final JavaPairRDD mapRDD = SparkHelper.executeMap(graphRDD, mapReduce, newApacheConfiguration);
@@ -191,7 +191,7 @@ public final class SparkGraphComputer implements GraphComputer {
sparkContext.close();
// update runtime and return the newly computed graph
finalMemory.setRuntime(System.currentTimeMillis() - startTime);
- return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph), finalMemory.asImmutable());
+ return new DefaultComputerResult(HadoopHelper.getOutputGraph(this.hadoopGraph, null != this.vertexProgram), finalMemory.asImmutable());
}
}
);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SApacheConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SApacheConfiguration.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SApacheConfiguration.java
deleted file mode 100644
index dbd468a..0000000
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/util/SApacheConfiguration.java
+++ /dev/null
@@ -1,70 +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.spark.util;
-
-import org.apache.commons.configuration.AbstractConfiguration;
-import org.apache.commons.configuration.Configuration;
-import org.apache.commons.configuration.ConfigurationUtils;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class SApacheConfiguration extends AbstractConfiguration implements Serializable {
-
- private final Map<String, Object> configurations = new HashMap<>();
-
- public SApacheConfiguration() {
- this.setDelimiterParsingDisabled(true); // gets me everytime (what a stupid default behavior)
- }
-
- public SApacheConfiguration(final Configuration configuration) {
- this();
- ConfigurationUtils.copy(configuration, this);
- }
-
- @Override
- protected void addPropertyDirect(final String key, final Object value) {
- this.configurations.put(key, value);
- }
-
- @Override
- public boolean isEmpty() {
- return this.configurations.isEmpty();
- }
-
- @Override
- public boolean containsKey(final String key) {
- return this.configurations.containsKey(key);
- }
-
- @Override
- public Object getProperty(final String key) {
- return this.configurations.get(key);
- }
-
- @Override
- public Iterator<String> getKeys() {
- return this.configurations.keySet().iterator();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 5bd8465..478e805 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
@@ -146,7 +146,7 @@ public final class SparkHelper {
if (null != outputLocation) {
// map back to a <nullwritable,vertexwritable> stream for output
graphRDD.mapToPair(tuple -> new Tuple2<>(NullWritable.get(), new VertexWritable(tuple._2().asVertexPayload().getVertex())))
- .saveAsNewAPIHadoopFile(outputLocation + "/" + Constants.SYSTEM_G,
+ .saveAsNewAPIHadoopFile(outputLocation + "/" + Constants.HIDDEN_G,
NullWritable.class,
VertexWritable.class,
(Class<OutputFormat<NullWritable, VertexWritable>>) hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, OutputFormat.class));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
index d6fa24b..6e07a39 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MapReduceHelper.java
@@ -102,7 +102,7 @@ public final class MapReduceHelper {
job.setOutputFormatClass(newConfiguration.getClass(Constants.GREMLIN_HADOOP_MEMORY_OUTPUT_FORMAT, SequenceFileOutputFormat.class, OutputFormat.class)); // TODO: Make this configurable
// if there is no vertex program, then grab the graph from the input location
final Path graphPath = vertexProgramExists ?
- new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.SYSTEM_G) :
+ new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + Constants.HIDDEN_G) :
new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION));
Path memoryPath = new Path(newConfiguration.get(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION) + "/" + (reduceSort.isPresent() ? mapReduce.getMemoryKey() + "-temp" : mapReduce.getMemoryKey()));
if (FileSystem.get(newConfiguration).exists(memoryPath)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopConfiguration.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopConfiguration.java
index 0fb2e9f..4f899a0 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopConfiguration.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/HadoopConfiguration.java
@@ -18,57 +18,63 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure;
-import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-import org.apache.tinkerpop.gremlin.util.StreamFactory;
-import org.javatuples.Pair;
import java.io.Serializable;
+import java.util.HashMap;
import java.util.Iterator;
+import java.util.Map;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public class HadoopConfiguration extends BaseConfiguration implements Serializable, Iterable {
+public class HadoopConfiguration extends AbstractConfiguration implements Serializable {
+
+ private final Map<String, Object> properties = new HashMap<>();
public HadoopConfiguration() {
+ super();
+ super.setDelimiterParsingDisabled(true);
+ }
+ @Override
+ protected void addPropertyDirect(final String key, final Object value) {
+ this.properties.put(key, value);
}
public HadoopConfiguration(final Configuration configuration) {
+ this();
this.copy(configuration);
}
- public Class<InputFormat<NullWritable, VertexWritable>> getGraphInputFormat() {
- try {
- return (Class) Class.forName(this.getString(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT));
- } catch (final ClassNotFoundException e) {
- throw new RuntimeException(e.getMessage(), e);
- }
+ @Override
+ public boolean isEmpty() {
+ return this.properties.isEmpty();
}
- public void setGraphInputFormat(final Class<InputFormat<NullWritable, VertexWritable>> inputFormatClass) {
- this.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, inputFormatClass);
+ @Override
+ public boolean containsKey(final String key) {
+ return this.properties.containsKey(key);
}
- public Class<OutputFormat<NullWritable, VertexWritable>> getGraphOutputFormat() {
- try {
- return (Class) Class.forName(this.getString(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT));
- } catch (final ClassNotFoundException e) {
- throw new RuntimeException(e.getMessage(), e);
- }
+ @Override
+ public Object getProperty(final String key) {
+ return this.properties.get(key);
}
- public void setGraphOutputFormat(final Class<OutputFormat<NullWritable, VertexWritable>> outputFormatClass) {
- this.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, outputFormatClass);
+ @Override
+ public Iterator<String> getKeys() {
+ return this.properties.keySet().iterator();
}
+ ///////
+
public String getInputLocation() {
return this.getString(Constants.GREMLIN_HADOOP_INPUT_LOCATION);
}
@@ -85,20 +91,19 @@ public class HadoopConfiguration extends BaseConfiguration implements Serializab
this.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, outputLocation);
}
- public Class<? extends GraphComputer> getGraphComputer(final Class<? extends GraphComputer> defaultGraphComputer) {
- if (!this.containsKey(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER))
- return defaultGraphComputer;
- else {
- try {
- return (Class) Class.forName(this.getString(Constants.GREMLIN_HADOOP_DEFAULT_GRAPH_COMPUTER));
- } catch (final ClassNotFoundException e) {
- throw new RuntimeException(e.getMessage(), e);
- }
+ public Class<InputFormat<NullWritable, VertexWritable>> getGraphInputFormat() {
+ try {
+ return (Class) Class.forName(this.getString(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT));
+ } catch (final ClassNotFoundException e) {
+ throw new RuntimeException(e.getMessage(), e);
}
}
- @Override
- public Iterator iterator() {
- return StreamFactory.stream(this.getKeys()).map(k -> new Pair(k, this.getProperty(k))).iterator();
+ public Class<OutputFormat<NullWritable, VertexWritable>> getGraphOutputFormat() {
+ try {
+ return (Class) Class.forName(this.getString(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT));
+ } catch (final ClassNotFoundException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/InputOutputHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/InputOutputHelper.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/InputOutputHelper.java
index e08d90e..f4f5d29 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/InputOutputHelper.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/InputOutputHelper.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io;
+import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONInputFormat;
@@ -27,34 +28,41 @@ import org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptOutputFormat;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
public class InputOutputHelper {
+ private static Map<Class<? extends InputFormat<NullWritable, VertexWritable>>, Class<? extends OutputFormat<NullWritable, VertexWritable>>> INPUT_TO_OUTPUT_CACHE = new ConcurrentHashMap<>();
+ private static Map<Class<? extends OutputFormat<NullWritable, VertexWritable>>, Class<? extends InputFormat<NullWritable, VertexWritable>>> OUTPUT_TO_INPUT_CACHE = new ConcurrentHashMap<>();
+
+ static {
+ INPUT_TO_OUTPUT_CACHE.put(GryoInputFormat.class, GryoOutputFormat.class);
+ INPUT_TO_OUTPUT_CACHE.put(GraphSONInputFormat.class, GraphSONOutputFormat.class);
+ INPUT_TO_OUTPUT_CACHE.put(ScriptInputFormat.class, ScriptOutputFormat.class);
+ //
+ OUTPUT_TO_INPUT_CACHE.put(GryoOutputFormat.class, GryoInputFormat.class);
+ OUTPUT_TO_INPUT_CACHE.put(GraphSONOutputFormat.class, GraphSONInputFormat.class);
+ OUTPUT_TO_INPUT_CACHE.put(ScriptOutputFormat.class, ScriptInputFormat.class);
+ }
+
private InputOutputHelper() {
}
- public static Class<? extends InputFormat> getInputFormat(final Class<? extends OutputFormat> outputFormat) {
- if (outputFormat.equals(GryoOutputFormat.class))
- return GryoInputFormat.class;
- else if (outputFormat.equals(GraphSONOutputFormat.class))
- return GraphSONInputFormat.class;
- else if (outputFormat.equals(ScriptOutputFormat.class))
- return ScriptInputFormat.class;
- else
- throw new IllegalArgumentException("The provided output format does not have a known input format: " + outputFormat.getCanonicalName());
+ public static Class<? extends InputFormat> getInputFormat(final Class<? extends OutputFormat<NullWritable, VertexWritable>> outputFormat) {
+ return OUTPUT_TO_INPUT_CACHE.get(outputFormat);
+ }
+
+ public static Class<? extends OutputFormat> getOutputFormat(final Class<? extends InputFormat<NullWritable, VertexWritable>> inputFormat) {
+ return INPUT_TO_OUTPUT_CACHE.get(inputFormat);
}
- public static Class<? extends OutputFormat> getOutputFormat(final Class<? extends InputFormat> inputFormat) {
- if (inputFormat.equals(GryoInputFormat.class))
- return GryoOutputFormat.class;
- else if (inputFormat.equals(GraphSONInputFormat.class))
- return GraphSONOutputFormat.class;
- else if (inputFormat.equals(ScriptInputFormat.class))
- return ScriptOutputFormat.class;
- else
- throw new IllegalArgumentException("The provided input format does not have a known output format: " + inputFormat.getCanonicalName());
+ public static void registerInputOutputPair(final Class<? extends InputFormat<NullWritable, VertexWritable>> inputFormat, final Class<? extends OutputFormat<NullWritable, VertexWritable>> outputFormat) {
+ INPUT_TO_OUTPUT_CACHE.put(inputFormat, outputFormat);
+ OUTPUT_TO_INPUT_CACHE.put(outputFormat, inputFormat);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 2535932..f6480c2 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
@@ -31,7 +31,6 @@ import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONReader;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
import java.io.ByteArrayInputStream;
import java.io.IOException;
@@ -61,16 +60,13 @@ public class GraphSONRecordReader extends RecordReader<NullWritable, VertexWrita
if (!this.lineRecordReader.nextKeyValue())
return false;
- final TinkerGraph g = TinkerGraph.open();
- final Function<DetachedVertex, Vertex> vertexMaker = detachedVertex -> DetachedVertex.addTo(g, detachedVertex);
- final Function<DetachedEdge, Edge> edgeMaker = detachedEdge -> DetachedEdge.addTo(g, detachedEdge);
- final TinkerVertex vertex;
+ final TinkerGraph gLocal = TinkerGraph.open();
+ 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())) {
- vertex = (TinkerVertex) GRAPHSON_READER.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker);
+ this.vertexWritable.set(GRAPHSON_READER.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker));
+ return true;
}
-
- this.vertexWritable.set(vertex);
- return true;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 d120a6f..83e364d 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
@@ -18,8 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
@@ -29,18 +27,39 @@ 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.structure.io.VertexWritable;
+import org.apache.tinkerpop.gremlin.structure.Direction;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoMapper;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoReader;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
+import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.util.function.Function;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable> {
- private VertexStreamIterator vertexStreamIterator;
private FSDataInputStream inputStream;
private static final byte[] PATTERN = GryoMapper.build().create().getVersionedHeader();
+ // this is VertexTerminator's long terminal 4185403236219066774L as an array of positive int's
+ private static final int[] TERMINATOR = new int[]{58, 21, 138, 17, 112, 155, 153, 150};
+
+ private final GryoReader gryoReader = GryoReader.build().create();
+ private final VertexWritable vertexWritable = new VertexWritable();
+
+ private long currentLength = 0;
+ private long splitLength;
public GryoRecordReader() {
}
@@ -57,8 +76,7 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
// open the file and seek to the start of the split
this.inputStream = file.getFileSystem(job).open(split.getPath());
this.inputStream.seek(start);
- final long newStart = seekToHeader(this.inputStream, start);
- this.vertexStreamIterator = new VertexStreamIterator(this.inputStream, split.getLength() - (newStart - start));
+ this.splitLength = split.getLength() - (seekToHeader(this.inputStream, start) - start);
}
private static long seekToHeader(final FSDataInputStream inputStream, final long start) throws IOException {
@@ -88,7 +106,36 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
@Override
public boolean nextKeyValue() throws IOException {
- return this.vertexStreamIterator.hasNext();
+ final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ long currentVertexLength = 0;
+ int terminatorLocation = 0;
+ while (true) {
+ final int currentByte = this.inputStream.read();
+ if (-1 == currentByte) {
+ if (currentVertexLength > 0)
+ throw new IllegalStateException("Remainder of stream exhausted without matching a vertex");
+ else
+ return false;
+ }
+ this.currentLength++;
+ currentVertexLength++;
+ output.write(currentByte);
+
+ if (currentByte == TERMINATOR[terminatorLocation])
+ terminatorLocation++;
+ else
+ terminatorLocation = 0;
+
+ if (terminatorLocation >= TERMINATOR.length) {
+ final Graph gLocal = TinkerGraph.open();
+ 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));
+ return true;
+ }
+ }
+ }
}
@Override
@@ -98,12 +145,15 @@ public class GryoRecordReader extends RecordReader<NullWritable, VertexWritable>
@Override
public VertexWritable getCurrentValue() {
- return this.vertexStreamIterator.next();
+ return this.vertexWritable;
}
@Override
public float getProgress() throws IOException {
- return this.vertexStreamIterator.getProgress();
+ if (0 == this.currentLength || 0 == this.splitLength)
+ return 0.0f;
+ else
+ return (float) this.currentLength / (float) this.splitLength;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 5558eaf..5e8f8a0 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
@@ -43,7 +43,7 @@ public class GryoRecordWriter extends RecordWriter<NullWritable, VertexWritable>
@Override
public void write(final NullWritable key, final VertexWritable vertex) throws IOException {
if (null != vertex) {
- GRYO_WRITER.writeVertex(out, vertex.get(), Direction.BOTH);
+ GRYO_WRITER.writeVertex(this.out, vertex.get(), Direction.BOTH);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIterator.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIterator.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIterator.java
deleted file mode 100644
index 7fb69ee..0000000
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIterator.java
+++ /dev/null
@@ -1,138 +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.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.Edge;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoReader;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedEdge;
-import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Iterator;
-import java.util.function.Function;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class VertexStreamIterator implements Iterator<VertexWritable> {
-
- // this is VertexTerminator's long terminal 4185403236219066774L as an array of positive int's
- private static final int[] TERMINATOR = new int[]{58, 21, 138, 17, 112, 155, 153, 150};
-
- private final InputStream inputStream;
- private final ByteArrayOutputStream output = new ByteArrayOutputStream();
- private final GryoReader gryoReader = GryoReader.build().create();
-
- private int currentByte;
- private long currentTotalLength = 0;
- private Vertex currentVertex;
- private final VertexWritable vertexWritable = new VertexWritable();
- private final long maxLength;
-
- public VertexStreamIterator(final InputStream inputStream, final long maxLength) {
- this.inputStream = inputStream;
- this.maxLength = maxLength;
- }
-
- public float getProgress() {
- if (0 == this.currentTotalLength || 0 == this.maxLength)
- return 0.0f;
- else if (this.currentTotalLength >= this.maxLength || this.maxLength == Long.MAX_VALUE)
- return 1.0f;
- else
- return (float) this.currentTotalLength / (float) this.maxLength;
-
- }
-
- @Override
- public boolean hasNext() {
- if (this.currentTotalLength >= this.maxLength) // gone beyond the split boundary
- return false;
- if (null != this.currentVertex)
- return true;
- else if (-1 == this.currentByte)
- return false;
- else {
- try {
- this.currentVertex = advanceToNextVertex();
- return null != this.currentVertex;
- } catch (final IOException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- }
- }
-
- @Override
- public VertexWritable next() {
- try {
- if (null == this.currentVertex) {
- if (this.hasNext()) {
- this.vertexWritable.set(this.currentVertex);
- return this.vertexWritable;
- } else
- throw new IllegalStateException("There are no more vertices in this split");
- } else {
- this.vertexWritable.set(this.currentVertex);
- return this.vertexWritable;
- }
- } finally {
- this.currentVertex = null;
- this.output.reset();
- }
- }
-
- private final Vertex advanceToNextVertex() throws IOException {
- long currentVertexLength = 0;
- int terminatorLocation = 0;
- while (true) {
- this.currentByte = this.inputStream.read();
- if (-1 == this.currentByte) {
- if (currentVertexLength > 0)
- throw new IllegalStateException("Remainder of stream exhausted without matching a vertex");
- else
- return null;
- }
- this.currentTotalLength++;
- currentVertexLength++;
- this.output.write(this.currentByte);
-
- if (this.currentByte == TERMINATOR[terminatorLocation])
- terminatorLocation++;
- else
- terminatorLocation = 0;
-
- if (terminatorLocation >= TERMINATOR.length) {
- final Graph gLocal = TinkerGraph.open();
- 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.output.toByteArray())) {
- return this.gryoReader.readVertex(in, Direction.BOTH, vertexMaker, edgeMaker);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/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 26a5f0d..075d903 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
@@ -18,9 +18,10 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.util;
+import org.apache.commons.configuration.BaseConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
-import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.InputOutputHelper;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -30,11 +31,15 @@ public final class HadoopHelper {
private HadoopHelper() {
}
- public static HadoopGraph getOutputGraph(final HadoopGraph hadoopGraph) {
+ public static HadoopGraph getOutputGraph(final HadoopGraph hadoopGraph, boolean hasVertexProgram) {
final BaseConfiguration newConfiguration = new BaseConfiguration();
newConfiguration.copy(hadoopGraph.configuration());
- if (hadoopGraph.configuration().containsKey(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION)) {
- newConfiguration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, hadoopGraph.configuration().getOutputLocation() + "/" + Constants.SYSTEM_G);
+ if (hasVertexProgram) {
+ 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() + "_");
+
+ } 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/8798d828/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 5e6704d..e1a661a 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
@@ -84,7 +84,7 @@ public class GryoRecordReaderWriterTest {
assertEquals(Integer.class, vertex.id().getClass());
Object value = vertex.property("name");
- if (null != value && ((Property) value).value().equals("SUGAR MAGNOLIA")) {
+ if (((Property) value).value().equals("SUGAR MAGNOLIA")) {
foundKeyValue = true;
assertEquals(92, IteratorUtils.count(vertex.edges(Direction.OUT)));
assertEquals(77, IteratorUtils.count(vertex.edges(Direction.IN)));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/8798d828/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIteratorTest.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIteratorTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIteratorTest.java
deleted file mode 100644
index 4c1ded6..0000000
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/VertexStreamIteratorTest.java
+++ /dev/null
@@ -1,99 +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.structure.io.gryo;
-
-import org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph.GiraphComputeVertex;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
-import org.apache.tinkerpop.gremlin.structure.Direction;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoWriter;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.FileInputStream;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class VertexStreamIteratorTest {
-
- @Test
- @Ignore
- public void testStuff() throws Exception {
- FileInputStream input = new FileInputStream("/tmp/100K.gio");
- VertexStreamIterator iterator = new VertexStreamIterator(input, 15542136);
- int counter = 0;
- long time = System.currentTimeMillis();
- while (iterator.hasNext()) {
- if (++counter % 1000 == 0) {
- System.out.println("Read vertices: " + counter + "[ms:" + (System.currentTimeMillis() - time) + "]");
- System.out.println(iterator.getProgress() + " -- progress");
- time = System.currentTimeMillis();
- }
- iterator.next();
- // final ByteArrayOutputStream out = new ByteArrayOutputStream();
- //System.out.println("HERE:" +IteratorUtils.count(vertex.get().edges(Direction.BOTH)));
- // final GiraphComputeVertex vertex = new GiraphComputeVertex(iterator.next());
- //System.out.println(vertex + "!!!!");
- // final DataOutputStream output = new DataOutputStream(out);
- // vertex.getValue().write(output);
- //output.flush();
- //System.out.println("!!!" + out.size());
- // final VertexWritable v = new VertexWritable();
- // final ByteArrayInputStream inputStream = new ByteArrayInputStream(out.toByteArray());
- //v.readFields(new DataInputStream(inputStream));
- }
- }
-
- @Test
- public void testAll() throws Exception {
- Graph g = TinkerFactory.createClassic();
- try (final ByteArrayOutputStream os = new ByteArrayOutputStream()) {
- final GryoWriter writer = GryoWriter.build().create();
- writer.writeVertices(os, g.traversal().V(), Direction.BOTH);
- final AtomicInteger called = new AtomicInteger(0);
- VertexStreamIterator vsi = new VertexStreamIterator(new ByteArrayInputStream(os.toByteArray()), Long.MAX_VALUE);
-
- boolean found = false;
- while (vsi.hasNext()) {
- Vertex v = vsi.next().get();
- String name = v.<String>property("name").value();
- if (name.equals("ripple")) {
- found = true;
- assertEquals(1, IteratorUtils.count(v.vertices(Direction.IN)));
- assertEquals(0, IteratorUtils.count(v.vertices(Direction.OUT)));
- }
- called.incrementAndGet();
- }
- assertTrue(found);
- assertEquals(IteratorUtils.count(g.vertices()), called.get());
- }
- }
-}