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/09/10 20:07:13 UTC
incubator-tinkerpop git commit: Support for Apache Hadoop 2.7.1,
Apache Spark 1.5.0, and Apache Giraph 1.1.0. Still some cleanup work to do,
but this is the big kahunna for shurzies.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master 4b32fe2b2 -> a384d554e
Support for Apache Hadoop 2.7.1, Apache Spark 1.5.0, and Apache Giraph 1.1.0. Still some cleanup work to do, but this is the big kahunna for shurzies.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/a384d554
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/a384d554
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/a384d554
Branch: refs/heads/master
Commit: a384d554e24f5fd9b34632b874d3fd2a50298aff
Parents: 4b32fe2
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Thu Sep 10 12:07:34 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Thu Sep 10 12:07:34 2015 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 2 +
giraph-gremlin/pom.xml | 58 ++++++++++++-----
.../process/computer/GiraphComputation.java | 47 ++++++++++++++
.../process/computer/GiraphComputeVertex.java | 15 +----
.../process/computer/GiraphGraphComputer.java | 3 +-
.../process/computer/GiraphMessageCombiner.java | 3 +-
.../process/computer/GiraphMessenger.java | 8 ++-
.../process/computer/GiraphWorkerContext.java | 4 +-
.../process/computer/MemoryAggregator.java | 2 +-
.../computer/io/GiraphVertexInputFormat.java | 9 ++-
.../groovy/plugin/HadoopGremlinPluginTest.java | 8 +--
.../computer/HadoopGiraphGraphProvider.java | 4 +-
hadoop-gremlin/pom.xml | 67 +++++++++++++++++---
.../tinkerpop/gremlin/hadoop/Constants.java | 2 +-
.../process/computer/util/MapReduceHelper.java | 4 +-
.../structure/hdfs/HadoopElementIterator.java | 21 +++---
.../structure/io/ObjectWritableIterator.java | 12 ++--
.../structure/io/RecordReaderWriterTest.java | 22 +++++--
pom.xml | 2 +-
spark-gremlin/pom.xml | 61 ++++++++++++------
.../process/computer/SparkGraphComputer.java | 4 +-
.../groovy/plugin/HadoopGremlinPluginTest.java | 8 +--
.../computer/HadoopSparkGraphProvider.java | 2 +-
.../process/computer/io/InputOutputRDDTest.java | 2 +-
.../spark/process/computer/io/InputRDDTest.java | 2 +-
.../process/computer/io/OutputRDDTest.java | 2 +-
26 files changed, 263 insertions(+), 111 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 8c79dbb..b93fcc2 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,8 @@ image::https://raw.githubusercontent.com/apache/incubator-tinkerpop/master/docs/
TinkerPop 3.1.0 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Bumped to Apache Hadoop 2.7.1.
+* Bumped to Apache Giraph 1.1.0.
* Bumped to Apache Spark 1.5.0.
* Split Hadoop-Gremlin apart such there is now `hadoop-gremlin`, `spark-gremlin`, and `giraph-gremlin` (and respective `GremlinPlugins`).
* Added `LambdaCollectingBarrierStep` which generalizes `NoOpBarrierStep` and allows for `barrier(normSack)`-type operations.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/giraph-gremlin/pom.xml b/giraph-gremlin/pom.xml
index c5e1a8e..1e2fb76 100644
--- a/giraph-gremlin/pom.xml
+++ b/giraph-gremlin/pom.xml
@@ -40,18 +40,58 @@ limitations under the License.
<groupId>org.apache.tinkerpop</groupId>
<artifactId>hadoop-gremlin</artifactId>
<version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.thoughtworks.paranamer</groupId>
+ <artifactId>paranamer</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
- <!-- GIRAPH GRAPH COMPUTER -->
+ <!-- GIRAPH -->
<dependency>
<groupId>org.apache.giraph</groupId>
<artifactId>giraph-core</artifactId>
- <version>1.0.0</version>
+ <version>1.1.0-hadoop2</version>
<exclusions>
<!-- self conflicts -->
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
+ <!-- conflicts with hadoop-gremlin -->
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ </exclusion>
+ <!-- conflicts with gremlin-groovy -->
+ <exclusion>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </exclusion>
<!-- conflicts with gremlin-core -->
<exclusion>
<groupId>org.slf4j</groupId>
@@ -68,12 +108,6 @@ limitations under the License.
</exclusion>
</exclusions>
</dependency>
- <!-- consistent dependencies -->
- <dependency>
- <groupId>commons-httpclient</groupId>
- <artifactId>commons-httpclient</artifactId>
- <version>3.1</version>
- </dependency>
<!-- TEST -->
<dependency>
<groupId>org.apache.tinkerpop</groupId>
@@ -100,12 +134,6 @@ limitations under the License.
<scope>test</scope>
</dependency>
</dependencies>
- <repositories>
- <repository>
- <id>hyracks-releases</id>
- <url>http://obelix.ics.uci.edu/nexus/content/groups/hyracks-public-releases/</url>
- </repository>
- </repositories>
<build>
<directory>${basedir}/target</directory>
<finalName>${project.artifactId}-${project.version}</finalName>
@@ -174,7 +202,7 @@ limitations under the License.
<configuration>
<archive>
<manifestEntries>
- <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-core:1.2.1
+ <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-client:2.7.1
</Gremlin-Plugin-Dependencies>
</manifestEntries>
</archive>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
new file mode 100644
index 0000000..bb3c21c
--- /dev/null
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputation.java
@@ -0,0 +1,47 @@
+/*
+ * 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.giraph.process.computer;
+
+import org.apache.giraph.graph.BasicComputation;
+import org.apache.giraph.graph.Vertex;
+import org.apache.hadoop.io.NullWritable;
+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 java.io.IOException;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class GiraphComputation extends BasicComputation<ObjectWritable, VertexWritable, NullWritable, ObjectWritable> {
+
+ public GiraphComputation() {
+ }
+
+ @Override
+ public void compute(final Vertex<ObjectWritable, VertexWritable, NullWritable> vertex, final Iterable<ObjectWritable> messages) throws IOException {
+ final GiraphWorkerContext workerContext = this.getWorkerContext();
+ final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
+ vertexProgram.execute(ComputerGraph.vertexProgram(vertex.getValue().get(), vertexProgram), workerContext.getMessenger((GiraphComputeVertex) vertex, this, messages.iterator()), workerContext.getMemory());
+ workerContext.getVertexProgramPool().offer(vertexProgram);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputeVertex.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputeVertex.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputeVertex.java
index b8de43a..1f14383 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputeVertex.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphComputeVertex.java
@@ -18,17 +18,15 @@
*/
package org.apache.tinkerpop.gremlin.giraph.process.computer;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.graph.DefaultVertex;
import org.apache.hadoop.io.NullWritable;
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;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class GiraphComputeVertex extends Vertex<ObjectWritable, VertexWritable, NullWritable, ObjectWritable> {
+public final class GiraphComputeVertex extends DefaultVertex<ObjectWritable, VertexWritable, NullWritable> {
public GiraphComputeVertex() {
}
@@ -37,14 +35,5 @@ public final class GiraphComputeVertex extends Vertex<ObjectWritable, VertexWrit
final VertexWritable newWritable = new VertexWritable();
newWritable.set(vertexWritable.get());
this.initialize(new ObjectWritable<>(newWritable.get().id()), newWritable, EmptyOutEdges.instance());
-
- }
-
- @Override
- public void compute(final Iterable<ObjectWritable> messages) {
- final GiraphWorkerContext workerContext = (GiraphWorkerContext) this.getWorkerContext();
- final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
- 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/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
index 7e87df4..ce718b3 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphGraphComputer.java
@@ -71,6 +71,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
configuration.getKeys().forEachRemaining(key -> this.giraphConfiguration.set(key, configuration.getProperty(key).toString()));
this.giraphConfiguration.setMasterComputeClass(GiraphMemory.class);
this.giraphConfiguration.setVertexClass(GiraphComputeVertex.class);
+ this.giraphConfiguration.setComputationClass(GiraphComputation.class);
this.giraphConfiguration.setWorkerContextClass(GiraphWorkerContext.class);
this.giraphConfiguration.setOutEdgesClass(EmptyOutEdges.class);
this.giraphConfiguration.setClass(GiraphConstants.VERTEX_ID_CLASS.getKey(), ObjectWritable.class, ObjectWritable.class);
@@ -87,7 +88,7 @@ public final class GiraphGraphComputer extends AbstractHadoopGraphComputer imple
final BaseConfiguration apacheConfiguration = new BaseConfiguration();
vertexProgram.storeState(apacheConfiguration);
ConfUtil.mergeApacheIntoHadoopConfiguration(apacheConfiguration, this.giraphConfiguration);
- this.vertexProgram.getMessageCombiner().ifPresent(combiner -> this.giraphConfiguration.setCombinerClass(GiraphMessageCombiner.class));
+ this.vertexProgram.getMessageCombiner().ifPresent(combiner -> this.giraphConfiguration.setMessageCombinerClass(GiraphMessageCombiner.class));
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessageCombiner.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessageCombiner.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessageCombiner.java
index 144d553..4d725b2 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessageCombiner.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessageCombiner.java
@@ -19,7 +19,6 @@
package org.apache.tinkerpop.gremlin.giraph.process.computer;
import org.apache.commons.configuration.Configuration;
-import org.apache.giraph.combiner.Combiner;
import org.apache.giraph.conf.ImmutableClassesGiraphConfigurable;
import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
@@ -31,7 +30,7 @@ import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
*/
-public final class GiraphMessageCombiner extends Combiner<ObjectWritable, ObjectWritable> implements ImmutableClassesGiraphConfigurable {
+public final class GiraphMessageCombiner extends org.apache.giraph.combiner.MessageCombiner<ObjectWritable, ObjectWritable> implements ImmutableClassesGiraphConfigurable {
private MessageCombiner messageCombiner;
private ImmutableClassesGiraphConfiguration configuration;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessenger.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessenger.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessenger.java
index ed93cf1..2b9ce4f 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessenger.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphMessenger.java
@@ -38,10 +38,12 @@ import java.util.Iterator;
public final class GiraphMessenger<M> implements Messenger<M> {
private GiraphComputeVertex giraphComputeVertex;
+ private GiraphComputation giraphComputation;
private Iterator<ObjectWritable<M>> messages;
- public GiraphMessenger(final GiraphComputeVertex giraphComputeVertex, final Iterator<ObjectWritable<M>> messages) {
+ public GiraphMessenger(final GiraphComputeVertex giraphComputeVertex, final GiraphComputation giraphComputation, final Iterator<ObjectWritable<M>> messages) {
this.giraphComputeVertex = giraphComputeVertex;
+ this.giraphComputation = giraphComputation;
this.messages = messages;
}
@@ -57,13 +59,13 @@ public final class GiraphMessenger<M> implements Messenger<M> {
final Traversal.Admin<Vertex, Edge> incidentTraversal = GiraphMessenger.setVertexStart(localMessageScope.getIncidentTraversal().get(), this.giraphComputeVertex.getValue().get());
final Direction direction = GiraphMessenger.getOppositeDirection(incidentTraversal);
incidentTraversal.forEachRemaining(edge ->
- this.giraphComputeVertex.sendMessage(
+ this.giraphComputation.sendMessage(
new ObjectWritable<>(edge.vertices(direction).next().id()),
new ObjectWritable<>(localMessageScope.getEdgeFunction().apply(message, edge))));
} else {
final MessageScope.Global globalMessageScope = (MessageScope.Global) messageScope;
globalMessageScope.vertices().forEach(vertex ->
- this.giraphComputeVertex.sendMessage(new ObjectWritable<>(vertex.id()), new ObjectWritable<>(message)));
+ this.giraphComputation.sendMessage(new ObjectWritable<>(vertex.id()), new ObjectWritable<>(message)));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphWorkerContext.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphWorkerContext.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphWorkerContext.java
index 6469d75..3f0b6fa 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphWorkerContext.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/GiraphWorkerContext.java
@@ -71,7 +71,7 @@ public final class GiraphWorkerContext extends WorkerContext {
return this.memory;
}
- public GiraphMessenger getMessenger(final GiraphComputeVertex giraphComputeVertex, final Iterator<ObjectWritable> messages) {
- return new GiraphMessenger(giraphComputeVertex, messages);
+ public GiraphMessenger getMessenger(final GiraphComputeVertex giraphComputeVertex, final GiraphComputation giraphComputation, final Iterator<ObjectWritable> messages) {
+ return new GiraphMessenger(giraphComputeVertex, giraphComputation, messages);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/MemoryAggregator.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/MemoryAggregator.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/MemoryAggregator.java
index b7da66a..1e56892 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/MemoryAggregator.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/MemoryAggregator.java
@@ -45,7 +45,7 @@ public final class MemoryAggregator implements Aggregator<Rule> {
@Override
public void setAggregatedValue(final Rule rule) {
- this.currentObject = rule.getObject();
+ this.currentObject = null == rule ? null : rule.getObject();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/io/GiraphVertexInputFormat.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/io/GiraphVertexInputFormat.java b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/io/GiraphVertexInputFormat.java
index 01989ff..e5407da 100644
--- a/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/io/GiraphVertexInputFormat.java
+++ b/giraph-gremlin/src/main/java/org/apache/tinkerpop/gremlin/giraph/process/computer/io/GiraphVertexInputFormat.java
@@ -18,8 +18,6 @@
*/
package org.apache.tinkerpop.gremlin.giraph.process.computer.io;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.giraph.io.VertexInputFormat;
import org.apache.giraph.io.VertexReader;
import org.apache.hadoop.conf.Configuration;
@@ -29,6 +27,8 @@ import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import java.io.IOException;
import java.util.List;
@@ -41,6 +41,11 @@ public final class GiraphVertexInputFormat extends VertexInputFormat {
private InputFormat<NullWritable, VertexWritable> hadoopGraphInputFormat;
@Override
+ public void checkInputSpecs(final Configuration configuration) {
+
+ }
+
+ @Override
public List<InputSplit> getSplits(final JobContext context, final int minSplitCountHint) throws IOException, InterruptedException {
this.constructor(context.getConfiguration());
return this.hadoopGraphInputFormat.getSplits(context);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/groovy/plugin/HadoopGremlinPluginTest.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/groovy/plugin/HadoopGremlinPluginTest.java b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/groovy/plugin/HadoopGremlinPluginTest.java
index d9fd95e..af24330 100644
--- a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/groovy/plugin/HadoopGremlinPluginTest.java
+++ b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/groovy/plugin/HadoopGremlinPluginTest.java
@@ -133,9 +133,9 @@ public class HadoopGremlinPluginTest extends AbstractGremlinTest {
this.remote.connect(Arrays.asList("graph", "g"));
Traversal<Vertex, String> traversal = (Traversal<Vertex, String>) this.remote.submit(Arrays.asList("g.V().hasLabel('person').group('m').by('age').by('name').out('knows').out('created').values('name')"));
AbstractGremlinProcessTest.checkResults(Arrays.asList("ripple", "lop"), traversal);
- assertTrue((Boolean) this.console.eval("hdfs.exists('hadoop-gremlin/target/test-output/m')"));
- assertTrue((Boolean) this.console.eval("hdfs.exists('hadoop-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "')"));
- final List<KeyValue<Integer, BulkSet<String>>> mList = IteratorUtils.asList(this.console.eval("hdfs.head('hadoop-gremlin/target/test-output/m',ObjectWritable)"));
+ assertTrue((Boolean) this.console.eval("hdfs.exists('giraph-gremlin/target/test-output/m')"));
+ assertTrue((Boolean) this.console.eval("hdfs.exists('giraph-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "')"));
+ final List<KeyValue<Integer, BulkSet<String>>> mList = IteratorUtils.asList(this.console.eval("hdfs.head('giraph-gremlin/target/test-output/m',ObjectWritable)"));
assertEquals(4, mList.size());
mList.forEach(keyValue -> {
if (keyValue.getKey().equals(29))
@@ -149,7 +149,7 @@ public class HadoopGremlinPluginTest extends AbstractGremlinTest {
else
throw new IllegalStateException("The provided key/value is unknown: " + keyValue);
});
- final List<KeyValue<MapReduce.NullObject, Traverser<String>>> traversersList = IteratorUtils.asList(this.console.eval("hdfs.head('hadoop-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "',ObjectWritable)"));
+ final List<KeyValue<MapReduce.NullObject, Traverser<String>>> traversersList = IteratorUtils.asList(this.console.eval("hdfs.head('giraph-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "',ObjectWritable)"));
assertEquals(2, traversersList.size());
traversersList.forEach(keyValue -> {
assertEquals(MapReduce.NullObject.instance(), keyValue.getKey());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
----------------------------------------------------------------------
diff --git a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
index 09547e8..1b3bdc4 100644
--- a/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
+++ b/giraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/giraph/process/computer/HadoopGiraphGraphProvider.java
@@ -111,7 +111,7 @@ public final class HadoopGiraphGraphProvider extends AbstractGraphProvider {
put(Graph.GRAPH, HadoopGraph.class.getName());
put(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, graphSONInput ? GraphSONInputFormat.class.getCanonicalName() : GryoInputFormat.class.getCanonicalName());
put(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, GryoOutputFormat.class.getCanonicalName());
- put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
+ put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "giraph-gremlin/target/test-output");
put(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
/// giraph configuration
put(GiraphConstants.MIN_WORKERS, 1);
@@ -123,7 +123,7 @@ public final class HadoopGiraphGraphProvider extends AbstractGraphProvider {
put(GiraphConstants.NUM_INPUT_THREADS.getKey(), 3);
put(GiraphConstants.NUM_COMPUTE_THREADS.getKey(), 3);
put(GiraphConstants.MAX_MASTER_SUPERSTEP_WAIT_MSECS.getKey(), TimeUnit.MINUTES.toMillis(60L));
- put("mapred.reduce.tasks", 4);
+ put("mapreduce.job.reduces", 4);
put("giraph.vertexOutputFormatThreadSafe", false);
put("giraph.numOutputThreads", 3);
}};
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/hadoop-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/pom.xml b/hadoop-gremlin/pom.xml
index 9161046..467397f 100644
--- a/hadoop-gremlin/pom.xml
+++ b/hadoop-gremlin/pom.xml
@@ -38,16 +38,70 @@ limitations under the License.
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-core</artifactId>
+ <artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
<exclusions>
<exclusion>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
</exclusion>
</exclusions>
<!--<scope>provided</scope>-->
</dependency>
+ <!-- consistent dependencies -->
+ <dependency>
+ <groupId>commons-codec</groupId>
+ <artifactId>commons-codec</artifactId>
+ <version>1.6</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ <version>1.1.3</version>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>16.0.1</version>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-mapper-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
+ <dependency>
+ <groupId>org.codehaus.jackson</groupId>
+ <artifactId>jackson-core-asl</artifactId>
+ <version>1.9.13</version>
+ </dependency>
<!-- TEST -->
<dependency>
<groupId>org.apache.tinkerpop</groupId>
@@ -136,13 +190,8 @@ limitations under the License.
<configuration>
<archive>
<manifestEntries>
- <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-core:1.2.1
+ <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-client:2.7.1
</Gremlin-Plugin-Dependencies>
- <!-- deletes the servlet-api jar from the path after install - causes conflicts -->
- <Gremlin-Plugin-Paths>servlet-api-2.5-6.1.14.jar=;servlet-api-2.5-20081211.jar=
- </Gremlin-Plugin-Paths>
- <Gremlin-Lib-Paths>servlet-api-2.5-6.1.14.jar=;servlet-api-2.5-20081211.jar=
- </Gremlin-Lib-Paths>
</manifestEntries>
</archive>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/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 17f3893..39cc8b1 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
@@ -47,7 +47,7 @@ public final class Constants {
public static final String HIDDEN_ITERATION = Graph.Hidden.hide("gremlin.hadoop.iteration");
public static final String GREMLIN_HADOOP_MAP_REDUCE_CLASS = "gremlin.hadoop.mapReduceClass";
- public static final String MAPRED_INPUT_DIR = "mapred.input.dir";
+ public static final String MAPREDUCE_INPUT_FILEINPUTFORMAT_INPUTDIR = "mapreduce.input.fileinputformat.inputdir";
// spark based constants
public static final String GREMLIN_HADOOP_GRAPH_INPUT_RDD = "gremlin.hadoop.graphInputRDD";
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/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 e0c590f..56af510 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
@@ -72,7 +72,7 @@ public final class MapReduceHelper {
final Optional<Comparator<?>> reduceSort = mapReduce.getReduceKeySort();
newConfiguration.setClass(Constants.GREMLIN_HADOOP_MAP_REDUCE_CLASS, mapReduce.getClass(), MapReduce.class);
- final Job job = new Job(newConfiguration, mapReduce.toString());
+ final Job job = Job.getInstance(newConfiguration, mapReduce.toString());
HadoopGraph.LOGGER.info(Constants.GREMLIN_HADOOP_JOB_PREFIX + mapReduce.toString());
job.setJarByClass(HadoopGraph.class);
if (mapSort.isPresent())
@@ -111,7 +111,7 @@ public final class MapReduceHelper {
// if there is a reduce sort, we need to run another identity MapReduce job
if (reduceSort.isPresent()) {
- final Job reduceSortJob = new Job(newConfiguration, "ReduceKeySort");
+ final Job reduceSortJob = Job.getInstance(newConfiguration, "ReduceKeySort");
reduceSortJob.setSortComparatorClass(ObjectWritableComparator.ObjectWritableReduceComparator.class);
reduceSortJob.setMapperClass(Mapper.class);
reduceSortJob.setReducerClass(Reducer.class);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/hdfs/HadoopElementIterator.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/hdfs/HadoopElementIterator.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/hdfs/HadoopElementIterator.java
index 1951b93..3602f85 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/hdfs/HadoopElementIterator.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/hdfs/HadoopElementIterator.java
@@ -18,21 +18,20 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.hdfs;
-import org.apache.tinkerpop.gremlin.hadoop.Constants;
-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;
-import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.tinkerpop.gremlin.hadoop.Constants;
+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;
+import org.apache.tinkerpop.gremlin.structure.Element;
import java.io.IOException;
import java.util.Iterator;
@@ -52,8 +51,8 @@ public abstract class HadoopElementIterator<E extends Element> implements Iterat
public HadoopElementIterator(final HadoopGraph graph, final InputFormat<NullWritable, VertexWritable> inputFormat, final Path path) throws IOException, InterruptedException {
this.graph = graph;
final Configuration configuration = ConfUtil.makeHadoopConfiguration(this.graph.configuration());
- for (final FileStatus status : FileSystem.get(configuration).listStatus(path, HiddenFileFilter.instance())) {
- this.readers.add(inputFormat.createRecordReader(new FileSplit(status.getPath(), 0, Integer.MAX_VALUE, new String[]{}), new TaskAttemptContext(configuration, new TaskAttemptID())));
+ for (final Path path2 : HDFSTools.getAllFilePaths(FileSystem.get(configuration), path, HiddenFileFilter.instance())) {
+ this.readers.add(inputFormat.createRecordReader(new FileSplit(path2, 0, Integer.MAX_VALUE, new String[]{}), new TaskAttemptContextImpl(configuration, new TaskAttemptID())));
}
}
@@ -63,8 +62,8 @@ public abstract class HadoopElementIterator<E extends Element> implements Iterat
if (this.graph.configuration().containsKey(Constants.GREMLIN_HADOOP_INPUT_LOCATION)) {
final Configuration configuration = ConfUtil.makeHadoopConfiguration(this.graph.configuration());
final InputFormat<NullWritable, VertexWritable> inputFormat = this.graph.configuration().getGraphInputFormat().getConstructor().newInstance();
- for (final FileStatus status : FileSystem.get(configuration).listStatus(new Path(graph.configuration().getInputLocation()), HiddenFileFilter.instance())) {
- this.readers.add(inputFormat.createRecordReader(new FileSplit(status.getPath(), 0, Integer.MAX_VALUE, new String[]{}), new TaskAttemptContext(configuration, new TaskAttemptID())));
+ for (final Path path : HDFSTools.getAllFilePaths(FileSystem.get(configuration), new Path(graph.configuration().getInputLocation()), HiddenFileFilter.instance())) {
+ this.readers.add(inputFormat.createRecordReader(new FileSplit(path, 0, Integer.MAX_VALUE, new String[]{}), new TaskAttemptContextImpl(configuration, new TaskAttemptID())));
}
}
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableIterator.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableIterator.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableIterator.java
index 8708033..60bf930 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableIterator.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableIterator.java
@@ -18,14 +18,14 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io;
-import org.apache.tinkerpop.gremlin.hadoop.structure.hdfs.HiddenFileFilter;
-import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
-import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
+import org.apache.tinkerpop.gremlin.hadoop.structure.hdfs.HDFSTools;
+import org.apache.tinkerpop.gremlin.hadoop.structure.hdfs.HiddenFileFilter;
+import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
+import org.apache.tinkerpop.gremlin.process.traversal.util.FastNoSuchElementException;
import java.io.IOException;
import java.util.Iterator;
@@ -44,8 +44,8 @@ public final class ObjectWritableIterator implements Iterator<KeyValue> {
public ObjectWritableIterator(final Configuration configuration, final Path path) throws IOException {
final FileSystem fs = FileSystem.get(configuration);
- for (final FileStatus status : fs.listStatus(path, HiddenFileFilter.instance())) {
- this.readers.add(new SequenceFile.Reader(fs, status.getPath(), configuration));
+ for (final Path path2 : HDFSTools.getAllFilePaths(fs, path, HiddenFileFilter.instance())) {
+ this.readers.add(new SequenceFile.Reader(configuration, SequenceFile.Reader.file(path2)));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java
index 96aab1c..eb6f119 100644
--- a/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java
+++ b/hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java
@@ -22,8 +22,15 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.tinkerpop.gremlin.TestHelper;
import org.apache.tinkerpop.gremlin.hadoop.HadoopGraphProvider;
@@ -44,6 +51,7 @@ import static org.junit.Assert.assertTrue;
/**
* @author Daniel Kuppitz (http://gremlin.guru)
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
*/
public abstract class RecordReaderWriterTest {
@@ -70,8 +78,8 @@ public abstract class RecordReaderWriterTest {
protected Configuration configure(final File outputDirectory) {
final Configuration configuration = new Configuration(false);
configuration.set("fs.file.impl", LocalFileSystem.class.getName());
- configuration.set("fs.default.name", "file:///");
- configuration.set("mapred.output.dir", "file:///" + outputDirectory.getAbsolutePath());
+ configuration.set("fs.defaultFS", "file:///");
+ configuration.set("mapreduce.output.fileoutputformat.outputdir", "file:///" + outputDirectory.getAbsolutePath());
return configuration;
}
@@ -90,7 +98,7 @@ public abstract class RecordReaderWriterTest {
final Optional<Class<? extends OutputFormat<NullWritable, VertexWritable>>> outFormatClass) throws Exception {
final InputFormat inputFormat = ReflectionUtils.newInstance(inputFormatClass, configuration);
- final TaskAttemptContext job = new TaskAttemptContext(configuration, new TaskAttemptID(UUID.randomUUID().toString(), 0, true, 0, 0));
+ final TaskAttemptContext job = new TaskAttemptContextImpl(configuration, new TaskAttemptID(UUID.randomUUID().toString(), 0, TaskType.MAP, 0, 0));
int vertexCount = 0;
int outEdgeCount = 0;
@@ -134,10 +142,10 @@ public abstract class RecordReaderWriterTest {
assertTrue(foundKeyValue);
if (null != writer) {
- writer.close(new TaskAttemptContext(configuration, job.getTaskAttemptID()));
+ writer.close(new TaskAttemptContextImpl(configuration, job.getTaskAttemptID()));
for (int i = 1; i < 10; i++) {
- final File outputDirectory = new File(new URL(configuration.get("mapred.output.dir")).toURI());
- final List<FileSplit> splits = generateFileSplits(new File(outputDirectory.getAbsoluteFile() + "/_temporary/" + job.getTaskAttemptID().getTaskID().toString().replace("task", "_attempt") + "_0" + "/part-m-00000"), i);
+ final File outputDirectory = new File(new URL(configuration.get("mapreduce.output.fileoutputformat.outputdir")).toURI());
+ final List<FileSplit> splits = generateFileSplits(new File(outputDirectory.getAbsoluteFile() + "/_temporary/0/_temporary/" + job.getTaskAttemptID().getTaskID().toString().replace("task", "attempt") + "_0" + "/part-m-00000"), i);
validateFileSplits(splits, configuration, inputFormatClass, Optional.empty());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 8025f05..32e7313 100644
--- a/pom.xml
+++ b/pom.xml
@@ -140,7 +140,7 @@ limitations under the License.
<skipIntegrationTests>true</skipIntegrationTests>
<skipPerformanceTests>true</skipPerformanceTests>
<slf4j.version>1.7.12</slf4j.version>
- <hadoop.version>1.2.1</hadoop.version>
+ <hadoop.version>2.7.1</hadoop.version>
<commons.configuration.version>1.10</commons.configuration.version>
<java.tuples.version>1.2</java.tuples.version>
<javadoc-plugin.version>2.10.1</javadoc-plugin.version>
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/pom.xml
----------------------------------------------------------------------
diff --git a/spark-gremlin/pom.xml b/spark-gremlin/pom.xml
index eb22f58..ae7c7dc 100644
--- a/spark-gremlin/pom.xml
+++ b/spark-gremlin/pom.xml
@@ -51,36 +51,56 @@
<version>${project.version}</version>
<exclusions>
<exclusion>
- <groupId>net.java.dev.jets3t</groupId>
- <artifactId>jets3t</artifactId>
+ <groupId>javax.servlet</groupId>
+ <artifactId>servlet-api</artifactId>
</exclusion>
<exclusion>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-server</artifactId>
</exclusion>
<exclusion>
<groupId>commons-net</groupId>
<artifactId>commons-net</artifactId>
</exclusion>
<exclusion>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>jetty</artifactId>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
</exclusion>
<exclusion>
- <groupId>org.mortbay.jetty</groupId>
- <artifactId>jsp-api-2.1</artifactId>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
</exclusion>
<exclusion>
- <groupId>com.sun.jersey</groupId>
- <artifactId>jersey-server</artifactId>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
</exclusion>
<exclusion>
- <groupId>com.sun.jersey</groupId>
- <artifactId>jersey-core</artifactId>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-math3</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.thoughtworks.paranamer</groupId>
+ <artifactId>paranamer</artifactId>
</exclusion>
</exclusions>
</dependency>
- <!-- SPARK GRAPH COMPUTER -->
+ <!-- SPARK -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
@@ -128,6 +148,10 @@
<artifactId>curator-framework</artifactId>
</exclusion>
<exclusion>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-recipes</artifactId>
+ </exclusion>
+ <exclusion>
<groupId>com.thoughtworks.paranamer</groupId>
<artifactId>paranamer</artifactId>
</exclusion>
@@ -162,6 +186,10 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-client</artifactId>
+ </exclusion>
<!-- lgpl conflicts -->
<exclusion>
<groupId>com.google.code.findbugs</groupId>
@@ -190,11 +218,6 @@
<artifactId>snappy-java</artifactId>
<version>1.1.1.7</version>
</dependency>
- <!--<dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-framework</artifactId>
- <version>2.4.0</version>
- </dependency>-->
<dependency>
<groupId>com.thoughtworks.paranamer</groupId>
<artifactId>paranamer</artifactId>
@@ -294,7 +317,7 @@
<configuration>
<archive>
<manifestEntries>
- <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-core:1.2.1
+ <Gremlin-Plugin-Dependencies>org.apache.hadoop:hadoop-client:2.7.1
</Gremlin-Plugin-Dependencies>
<!-- deletes the servlet-api jar from the path after install - causes conflicts -->
<Gremlin-Plugin-Paths>servlet-api-2.5-6.1.14.jar=;servlet-api-2.5-20081211.jar=
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
index f215806..4c9cec3 100644
--- a/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
+++ b/spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkGraphComputer.java
@@ -74,8 +74,8 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
if (FileInputFormat.class.isAssignableFrom(hadoopConfiguration.getClass(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputFormat.class))) {
try {
final String inputLocation = FileSystem.get(hadoopConfiguration).getFileStatus(new Path(hadoopConfiguration.get(Constants.GREMLIN_HADOOP_INPUT_LOCATION))).getPath().toString();
- apacheConfiguration.setProperty(Constants.MAPRED_INPUT_DIR, inputLocation);
- hadoopConfiguration.set(Constants.MAPRED_INPUT_DIR, inputLocation);
+ apacheConfiguration.setProperty(Constants.MAPREDUCE_INPUT_FILEINPUTFORMAT_INPUTDIR, inputLocation);
+ hadoopConfiguration.set(Constants.MAPREDUCE_INPUT_FILEINPUTFORMAT_INPUTDIR, inputLocation);
} catch (final IOException e) {
throw new IllegalStateException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/HadoopGremlinPluginTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/HadoopGremlinPluginTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/HadoopGremlinPluginTest.java
index 5931418..84e42d8 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/HadoopGremlinPluginTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/groovy/plugin/HadoopGremlinPluginTest.java
@@ -133,9 +133,9 @@ public class HadoopGremlinPluginTest extends AbstractGremlinTest {
this.remote.connect(Arrays.asList("graph", "g"));
Traversal<Vertex, String> traversal = (Traversal<Vertex, String>) this.remote.submit(Arrays.asList("g.V().hasLabel('person').group('m').by('age').by('name').out('knows').out('created').values('name')"));
AbstractGremlinProcessTest.checkResults(Arrays.asList("ripple", "lop"), traversal);
- assertTrue((Boolean) this.console.eval("hdfs.exists('hadoop-gremlin/target/test-output/m')"));
- assertTrue((Boolean) this.console.eval("hdfs.exists('hadoop-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "')"));
- final List<KeyValue<Integer, BulkSet<String>>> mList = IteratorUtils.asList(this.console.eval("hdfs.head('hadoop-gremlin/target/test-output/m',ObjectWritable)"));
+ assertTrue((Boolean) this.console.eval("hdfs.exists('spark-gremlin/target/test-output/m')"));
+ assertTrue((Boolean) this.console.eval("hdfs.exists('spark-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "')"));
+ final List<KeyValue<Integer, BulkSet<String>>> mList = IteratorUtils.asList(this.console.eval("hdfs.head('spark-gremlin/target/test-output/m',ObjectWritable)"));
assertEquals(4, mList.size());
mList.forEach(keyValue -> {
if (keyValue.getKey().equals(29))
@@ -149,7 +149,7 @@ public class HadoopGremlinPluginTest extends AbstractGremlinTest {
else
throw new IllegalStateException("The provided key/value is unknown: " + keyValue);
});
- final List<KeyValue<MapReduce.NullObject, Traverser<String>>> traversersList = IteratorUtils.asList(this.console.eval("hdfs.head('hadoop-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "',ObjectWritable)"));
+ final List<KeyValue<MapReduce.NullObject, Traverser<String>>> traversersList = IteratorUtils.asList(this.console.eval("hdfs.head('spark-gremlin/target/test-output/" + TraverserMapReduce.TRAVERSERS + "',ObjectWritable)"));
assertEquals(2, traversersList.size());
traversersList.forEach(keyValue -> {
assertEquals(MapReduce.NullObject.instance(), keyValue.getKey());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
index 21b14be..22aa3da 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/HadoopSparkGraphProvider.java
@@ -109,7 +109,7 @@ public final class HadoopSparkGraphProvider extends AbstractGraphProvider {
put(Graph.GRAPH, HadoopGraph.class.getName());
put(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, graphSONInput ? GraphSONInputFormat.class.getCanonicalName() : GryoInputFormat.class.getCanonicalName());
put(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, GryoOutputFormat.class.getCanonicalName());
- put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
+ put(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "spark-gremlin/target/test-output");
put(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
put("mapred.reduce.tasks", 4);
/// spark configuration
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputOutputRDDTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputOutputRDDTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputOutputRDDTest.java
index 8f35b54..6d1685e 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputOutputRDDTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputOutputRDDTest.java
@@ -44,7 +44,7 @@ public class InputOutputRDDTest {
configuration.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_RDD, ExampleInputRDD.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_RDD, ExampleOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "target/test-output");
configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
////////
Graph graph = GraphFactory.open(configuration);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputRDDTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputRDDTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputRDDTest.java
index 8ac9ba3..c07b4bb 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputRDDTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/InputRDDTest.java
@@ -44,7 +44,7 @@ public class InputRDDTest {
configuration.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_RDD, ExampleInputRDD.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_FORMAT, GryoOutputFormat.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "target/test-output");
configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
////////
Graph graph = GraphFactory.open(configuration);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/a384d554/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/OutputRDDTest.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/OutputRDDTest.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/OutputRDDTest.java
index 6a2c24c..2933af6 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/OutputRDDTest.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/io/OutputRDDTest.java
@@ -47,7 +47,7 @@ public class OutputRDDTest {
configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, HadoopSparkGraphProvider.PATHS.get("tinkerpop-modern.kryo"));
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, GryoInputFormat.class.getCanonicalName());
configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_OUTPUT_RDD, ExampleOutputRDD.class.getCanonicalName());
- configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "hadoop-gremlin/target/test-output");
+ configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, "target/test-output");
configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
////////
Graph graph = GraphFactory.open(configuration);