You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2015/05/08 20:00:20 UTC
incubator-tinkerpop git commit: VertexProgram.loadState() and
MapReduce.loadState() now take a Graph argument. This greatly simplifies
numerous areas of the code base around getting the graph to the OLAP job. So
much code was gutted that did backwards wo
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/master e99a57b23 -> 48c15ee28
VertexProgram.loadState() and MapReduce.loadState() now take a Graph argument. This greatly simplifies numerous areas of the code base around getting the graph to the OLAP job. So much code was gutted that did backwards workarounds to get to that information. Geez. Should have done this on day one.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/48c15ee2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/48c15ee2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/48c15ee2
Branch: refs/heads/master
Commit: 48c15ee281a4fcc866a30483463ef59b8ec7fac0
Parents: e99a57b
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Fri May 8 12:00:14 2015 -0600
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Fri May 8 12:00:14 2015 -0600
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../gremlin/process/computer/MapReduce.java | 9 ++++---
.../gremlin/process/computer/VertexProgram.java | 15 ++++++------
.../peerpressure/ClusterCountMapReduce.java | 3 ++-
.../ClusterPopulationMapReduce.java | 3 ++-
.../peerpressure/PeerPressureVertexProgram.java | 3 ++-
.../ranking/pagerank/PageRankMapReduce.java | 3 ++-
.../ranking/pagerank/PageRankVertexProgram.java | 3 ++-
.../traversal/TraversalVertexProgram.java | 12 ++++++----
.../traversal/step/map/ComputerResultStep.java | 2 +-
.../mapreduce/TraverserMapReduce.java | 5 ++--
.../util/AbstractVertexProgramBuilder.java | 5 ++--
.../process/traversal/step/map/GroupStep.java | 25 +++++++++-----------
.../step/sideEffect/AggregateStep.java | 5 ++--
.../sideEffect/GroupCountSideEffectStep.java | 5 ++--
.../step/sideEffect/GroupSideEffectStep.java | 7 +++---
.../traversal/step/sideEffect/StoreStep.java | 5 ++--
.../step/sideEffect/TreeSideEffectStep.java | 3 ++-
.../ranking/PageRankVertexProgramTest.java | 2 +-
.../groovy/plugin/HadoopRemoteAcceptor.java | 2 +-
.../hadoop/process/computer/HadoopCombine.java | 5 +++-
.../hadoop/process/computer/HadoopMap.java | 5 +++-
.../hadoop/process/computer/HadoopReduce.java | 6 +++--
.../computer/giraph/GiraphGraphComputer.java | 2 +-
.../process/computer/giraph/GiraphMemory.java | 6 ++++-
.../computer/giraph/GiraphMessageCombiner.java | 5 +++-
.../computer/giraph/GiraphWorkerContext.java | 5 +++-
.../process/computer/spark/SparkExecutor.java | 9 +++----
.../computer/spark/SparkGraphComputer.java | 2 +-
.../process/computer/util/MemoryMapReduce.java | 3 ++-
.../structure/io/ObjectWritableComparator.java | 7 ++++--
31 files changed, 105 insertions(+), 68 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 91b82c5..ecfa23c 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -25,6 +25,7 @@ image::http://www.tinkerpop.com/docs/current/images/gremlin-hindu.png[width=225]
TinkerPop 3.0.0.M9 (NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* `VertexProgram` and `MapReduce` now add a `Graph` parameter to `loadState(Graph,Configuration)`.
* Added `ScopingStrategy` which auto-scopes `select()` and `where()` so the language looks clean.
* Added `Scoping` as a marker interface to state that a step desires a particular `Scope`.
* `SelectStep`, `SelectOneStep`, and `WhereStep` support both `Scope.local` and `Scope.global` for `Map<String,Object>` or `Path` analysis, respectively.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/MapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/MapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/MapReduce.java
index 91e3afe..4852a9c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/MapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/MapReduce.java
@@ -19,6 +19,7 @@
package org.apache.tinkerpop.gremlin.process.computer;
import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import java.io.IOException;
@@ -69,9 +70,10 @@ public interface MapReduce<MK, MV, RK, RV, R> extends Cloneable {
* It is important that the state loaded from loadState() is identical to any state created from a constructor.
* For those GraphComputers that do not need to use Configurations to migrate state between JVMs, the constructor will only be used.
*
+ * @param graph the graph the MapReduce job will run against
* @param configuration the configuration to load the state of the MapReduce job from.
*/
- public default void loadState(final Configuration configuration) {
+ public default void loadState(final Graph graph, final Configuration configuration) {
}
@@ -208,16 +210,17 @@ public interface MapReduce<MK, MV, RK, RV, R> extends Cloneable {
* The class of the MapReduce is read from the {@link MapReduce#MAP_REDUCE} static configuration key.
* Once the MapReduce is constructed, {@link MapReduce#loadState} method is called with the provided configuration.
*
+ * @param graph The graph that the MapReduce job will run against
* @param configuration A configuration with requisite information to build a MapReduce
* @return the newly constructed MapReduce
*/
- public static <M extends MapReduce> M createMapReduce(final Configuration configuration) {
+ public static <M extends MapReduce> M createMapReduce(final Graph graph, final Configuration configuration) {
try {
final Class<M> mapReduceClass = (Class) Class.forName(configuration.getString(MAP_REDUCE));
final Constructor<M> constructor = mapReduceClass.getDeclaredConstructor();
constructor.setAccessible(true);
final M mapReduce = constructor.newInstance();
- mapReduce.loadState(configuration);
+ mapReduce.loadState(graph, configuration);
return mapReduce;
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
index 533ea60..4fd9e82 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/VertexProgram.java
@@ -20,6 +20,7 @@
package org.apache.tinkerpop.gremlin.process.computer;
import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import java.lang.reflect.Constructor;
@@ -62,12 +63,11 @@ public interface VertexProgram<M> extends Cloneable {
* When it is necessary to load the state of the VertexProgram, this method is called.
* This is typically required when the VertexProgram needs to be serialized to another machine.
* Note that what is loaded is simply the instance state, not any processed data.
- * It is possible for this method to be called for each and every vertex in the graph.
- * Thus, it is important to only store/load that state which is needed during execution in order to reduce startup time.
*
+ * @param graph the graph that the VertexProgram will run against
* @param configuration the configuration to load the state of the VertexProgram from.
*/
- public default void loadState(final Configuration configuration) {
+ public default void loadState(final Graph graph, final Configuration configuration) {
}
@@ -201,19 +201,20 @@ public interface VertexProgram<M> extends Cloneable {
/**
* A helper method to construct a {@link VertexProgram} given the content of the supplied configuration.
* The class of the VertexProgram is read from the {@link VertexProgram#VERTEX_PROGRAM} static configuration key.
- * Once the VertexProgram is constructed, {@link VertexProgram#loadState} method is called with the provided configuration.
+ * Once the VertexProgram is constructed, {@link VertexProgram#loadState} method is called with the provided graph and configuration.
*
+ * @param graph The graph that the vertex program will execute against
* @param configuration A configuration with requisite information to build a vertex program
* @param <V> The vertex program type
* @return the newly constructed vertex program
*/
- public static <V extends VertexProgram> V createVertexProgram(final Configuration configuration) {
+ public static <V extends VertexProgram> V createVertexProgram(final Graph graph, final Configuration configuration) {
try {
final Class<V> vertexProgramClass = (Class) Class.forName(configuration.getString(VERTEX_PROGRAM));
final Constructor<V> constructor = vertexProgramClass.getDeclaredConstructor();
constructor.setAccessible(true);
final V vertexProgram = constructor.newInstance();
- vertexProgram.loadState(configuration);
+ vertexProgram.loadState(graph, configuration);
return vertexProgram;
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
@@ -224,7 +225,7 @@ public interface VertexProgram<M> extends Cloneable {
public Builder configure(final Object... keyValues);
- public <P extends VertexProgram> P create();
+ public <P extends VertexProgram> P create(final Graph graph);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterCountMapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterCountMapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterCountMapReduce.java
index 5f625d5..4b391ed 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterCountMapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterCountMapReduce.java
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -57,7 +58,7 @@ public class ClusterCountMapReduce extends StaticMapReduce<MapReduce.NullObject,
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.memoryKey = configuration.getString(CLUSTER_COUNT_MEMORY_KEY, DEFAULT_MEMORY_KEY);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterPopulationMapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterPopulationMapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterPopulationMapReduce.java
index e172738..c87926c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterPopulationMapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/ClusterPopulationMapReduce.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.process.computer.clustering.peerpressure;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -54,7 +55,7 @@ public class ClusterPopulationMapReduce extends StaticMapReduce<Serializable, Lo
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.memoryKey = configuration.getString(CLUSTER_POPULATION_MEMORY_KEY, DEFAULT_MEMORY_KEY);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
index c9070dd..323dd15 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java
@@ -30,6 +30,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.MapHelper;
import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -75,7 +76,7 @@ public class PeerPressureVertexProgram extends StaticVertexProgram<Pair<Serializ
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.traversalSupplier = LambdaHolder.loadState(configuration, INCIDENT_TRAVERSAL_SUPPLIER);
if (null != this.traversalSupplier) {
this.voteScope = MessageScope.Local.of(this.traversalSupplier.get());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankMapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankMapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankMapReduce.java
index 861adaf..00e211d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankMapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankMapReduce.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.process.computer.ranking.pagerank;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Property;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -52,7 +53,7 @@ public class PageRankMapReduce extends StaticMapReduce<Object, Double, Object, D
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.memoryKey = configuration.getString(PAGE_RANK_MEMORY_KEY, DEFAULT_MEMORY_KEY);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
index cbd999c..96492b6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/pagerank/PageRankVertexProgram.java
@@ -30,6 +30,7 @@ import org.apache.tinkerpop.gremlin.process.computer.util.StaticVertexProgram;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -69,7 +70,7 @@ public class PageRankVertexProgram extends StaticVertexProgram<Double> {
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.traversalSupplier = LambdaHolder.loadState(configuration, INCIDENT_TRAVERSAL_SUPPLIER);
if (null != this.traversalSupplier) {
this.incidentMessageScope = MessageScope.Local.of(this.traversalSupplier.get());
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
index 4f630c6..7b4e8b0 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/TraversalVertexProgram.java
@@ -96,11 +96,12 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
* A helper method to yield a {@link Supplier} of {@link Traversal} from the {@link Configuration}.
* The supplier is either a {@link Class}, {@link org.apache.tinkerpop.gremlin.process.computer.util.ScriptEngineLambda}, or a direct Java8 lambda.
*
+ * @param graph the graph that the traversal will run against
* @param configuration The configuration containing the public static TRAVERSAL_SUPPLIER key.
* @return the traversal supplier in the configuration
*/
- public static Supplier<Traversal.Admin<?, ?>> getTraversalSupplier(final Configuration configuration) {
- return LambdaHolder.<Supplier<Traversal.Admin<?, ?>>>loadState(configuration, TraversalVertexProgram.TRAVERSAL_SUPPLIER).get();
+ public static Traversal.Admin<?, ?> getTraversal(final Graph graph, final Configuration configuration) {
+ return VertexProgram.<TraversalVertexProgram>createVertexProgram(graph, configuration).getTraversal();
}
public Traversal.Admin<?, ?> getTraversal() {
@@ -108,13 +109,16 @@ public final class TraversalVertexProgram implements VertexProgram<TraverserSet<
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.traversalSupplier = LambdaHolder.loadState(configuration, TRAVERSAL_SUPPLIER);
if (null == this.traversalSupplier) {
throw new IllegalArgumentException("The configuration does not have a traversal supplier");
}
this.traversal = this.traversalSupplier.get().get();
- if (!this.traversal.isLocked()) this.traversal.applyStrategies();
+ if (!this.traversal.isLocked()) {
+ this.traversal.setGraph(graph);
+ this.traversal.applyStrategies();
+ }
((ComputerResultStep) this.traversal.getEndStep()).byPass();
this.traversalMatrix = new TraversalMatrix<>(this.traversal);
for (final MapReducer<?, ?, ?, ?, ?> mapReducer : TraversalHelper.getStepsOfAssignableClassRecursively(MapReducer.class, this.traversal)) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
index 80e8364..a5e49c9 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/map/ComputerResultStep.java
@@ -76,7 +76,7 @@ public final class ComputerResultStep<S> extends AbstractStep<S, S> {
if (this.byPass) return this.starts.next();
if (this.first && null == this.computerResult) {
try {
- populateTraversers(this.graphComputer.program(TraversalVertexProgram.build().traversal(this.getTraversal(), !this.graphComputer.features().supportsDirectObjects()).create()).submit().get());
+ populateTraversers(this.graphComputer.program(TraversalVertexProgram.build().traversal(this.getTraversal(), !this.graphComputer.features().supportsDirectObjects()).create(this.graph)).submit().get());
} catch (final Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
index 3e38325..a691fd5 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/traversal/step/sideEffect/mapreduce/TraverserMapReduce.java
@@ -25,7 +25,6 @@ import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.step.ComparatorHolder;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.OrderGlobalStep;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.structure.Graph;
@@ -57,8 +56,8 @@ public final class TraverserMapReduce extends StaticMapReduce<Comparable, Object
}
@Override
- public void loadState(final Configuration configuration) {
- this.traversal = TraversalVertexProgram.getTraversalSupplier(configuration).get();
+ public void loadState(final Graph graph, final Configuration configuration) {
+ this.traversal = TraversalVertexProgram.getTraversal(graph, configuration);
final Step endStep = this.traversal.getEndStep().getPreviousStep(); // don't get the ComputerResultStep
this.comparator = Optional.ofNullable(endStep instanceof OrderGlobalStep ? new ChainedComparator<Comparable>(((OrderGlobalStep) endStep).getComparators()) : null);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/AbstractVertexProgramBuilder.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/AbstractVertexProgramBuilder.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/AbstractVertexProgramBuilder.java
index dd5cfe4..035b685 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/AbstractVertexProgramBuilder.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/util/AbstractVertexProgramBuilder.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.process.computer.util;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.tinkerpop.gremlin.structure.Graph;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -53,7 +54,7 @@ public abstract class AbstractVertexProgramBuilder<B extends VertexProgram.Build
}
@Override
- public <P extends VertexProgram> P create() {
- return (P) VertexProgram.createVertexProgram(this.configuration);
+ public <P extends VertexProgram> P create(final Graph graph) {
+ return (P) VertexProgram.createVertexProgram(graph, this.configuration);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
index 2dc1be6..3092c2f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/map/GroupStep.java
@@ -19,21 +19,22 @@
package org.apache.tinkerpop.gremlin.process.traversal.step.map;
import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.computer.KeyValue;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.traversal.TraversalVertexProgram;
-import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.ReducingBarrierStep;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
@@ -209,13 +210,9 @@ public final class GroupStep<S, K, V, R> extends ReducingBarrierStep<S, Map<K, R
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.groupStepId = configuration.getString(GROUP_BY_STEP_STEP_ID);
- final Traversal.Admin<?, ?> traversal = TraversalVertexProgram.getTraversalSupplier(configuration).get();
- if (!traversal.isLocked())
- traversal.applyStrategies(); // TODO: this is a scary error prone requirement, but only a problem for GroupStep
- final GroupStep groupStep = new TraversalMatrix<>(traversal).getStepById(this.groupStepId);
- this.reduceTraversal = groupStep.getReduceTraversal();
+ this.reduceTraversal = ((GroupStep) new TraversalMatrix<>(TraversalVertexProgram.getTraversal(graph, configuration)).getStepById(this.groupStepId)).getReduceTraversal();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateStep.java
index 3ce80a5..cf19d45 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/AggregateStep.java
@@ -34,6 +34,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequire
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.BulkSetSupplier;
@@ -130,9 +131,9 @@ public final class AggregateStep<S> extends CollectingBarrierStep<S> implements
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.sideEffectKey = configuration.getString(AGGREGATE_STEP_SIDE_EFFECT_KEY);
- this.collectionSupplier = TraversalVertexProgram.getTraversalSupplier(configuration).get().getSideEffects().<Collection>getRegisteredSupplier(this.sideEffectKey).orElse(BulkSet::new);
+ this.collectionSupplier = TraversalVertexProgram.getTraversal(graph, configuration).getSideEffects().<Collection>getRegisteredSupplier(this.sideEffectKey).orElse(BulkSet::new);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupCountSideEffectStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupCountSideEffectStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupCountSideEffectStep.java
index de88730..9b69884 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupCountSideEffectStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupCountSideEffectStep.java
@@ -33,6 +33,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.util.MapHelper;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.HashMapSupplier;
@@ -127,9 +128,9 @@ public final class GroupCountSideEffectStep<S, E> extends SideEffectStep<S> impl
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.sideEffectKey = configuration.getString(GROUP_COUNT_SIDE_EFFECT_STEP_SIDE_EFFECT_KEY);
- this.mapSupplier = TraversalVertexProgram.getTraversalSupplier(configuration).get().getSideEffects().<Map<E, Long>>getRegisteredSupplier(this.sideEffectKey).orElse(HashMap::new);
+ this.mapSupplier = TraversalVertexProgram.getTraversal(graph, configuration).getSideEffects().<Map<E, Long>>getRegisteredSupplier(this.sideEffectKey).orElse(HashMap::new);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
index 57fbc6f..6d469e7 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/GroupSideEffectStep.java
@@ -35,6 +35,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequire
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.function.HashMapSupplier;
@@ -187,12 +188,10 @@ public final class GroupSideEffectStep<S, K, V, R> extends SideEffectStep<S> imp
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.sideEffectKey = configuration.getString(GROUP_SIDE_EFFECT_STEP_SIDE_EFFECT_KEY);
this.groupStepId = configuration.getString(GROUP_SIDE_EFFECT_STEP_STEP_ID);
- final Traversal.Admin<?, ?> traversal = TraversalVertexProgram.getTraversalSupplier(configuration).get();
- if (!traversal.isLocked())
- traversal.applyStrategies(); // TODO: this is a scary error prone requirement, but only a problem for GroupStep
+ final Traversal.Admin<?, ?> traversal = TraversalVertexProgram.getTraversal(graph, configuration);
final GroupSideEffectStep groupSideEffectStep = new TraversalMatrix<>(traversal).getStepById(this.groupStepId);
this.reduceTraversal = groupSideEffectStep.getReduceTraversal();
this.mapSupplier = traversal.getSideEffects().<Map<K, R>>getRegisteredSupplier(this.sideEffectKey).orElse(HashMap::new);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/StoreStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/StoreStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/StoreStep.java
index 3c1a979..b721b85 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/StoreStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/StoreStep.java
@@ -33,6 +33,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.BulkSetSupplier;
@@ -128,9 +129,9 @@ public final class StoreStep<S> extends SideEffectStep<S> implements SideEffectC
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.sideEffectKey = configuration.getString(STORE_STEP_SIDE_EFFECT_KEY);
- this.collectionSupplier = TraversalVertexProgram.getTraversalSupplier(configuration).get().getSideEffects().<Collection>getRegisteredSupplier(this.sideEffectKey).orElse(BulkSet::new);
+ this.collectionSupplier = TraversalVertexProgram.getTraversal(graph, configuration).getSideEffects().<Collection>getRegisteredSupplier(this.sideEffectKey).orElse(BulkSet::new);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/TreeSideEffectStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/TreeSideEffectStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/TreeSideEffectStep.java
index c4f1288..176ac06 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/TreeSideEffectStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/step/sideEffect/TreeSideEffectStep.java
@@ -34,6 +34,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalRing;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalUtil;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.util.function.TreeSupplier;
@@ -138,7 +139,7 @@ public final class TreeSideEffectStep<S> extends SideEffectStep<S> implements Si
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.sideEffectKey = configuration.getString(TREE_SIDE_EFFECT_STEP_SIDE_EFFECT_KEY);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
index e7bcaf3..beb2824 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/computer/ranking/PageRankVertexProgramTest.java
@@ -38,7 +38,7 @@ public class PageRankVertexProgramTest extends AbstractGremlinProcessTest {
@Test
@LoadGraphWith(MODERN)
public void shouldExecutePageRank() throws Exception {
- final ComputerResult result = graph.compute(g.getGraphComputer().get().getClass()).program(PageRankVertexProgram.build().create()).submit().get();
+ final ComputerResult result = graph.compute(g.getGraphComputer().get().getClass()).program(PageRankVertexProgram.build().create(graph)).submit().get();
result.graph().traversal().V().forEachRemaining(v -> {
assertTrue(v.keys().contains("name"));
assertTrue(v.keys().contains(PageRankVertexProgram.PAGE_RANK));
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/groovy/plugin/HadoopRemoteAcceptor.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/groovy/plugin/HadoopRemoteAcceptor.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/groovy/plugin/HadoopRemoteAcceptor.java
index 93c2413..c3f14ff 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/groovy/plugin/HadoopRemoteAcceptor.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/groovy/plugin/HadoopRemoteAcceptor.java
@@ -100,7 +100,7 @@ public class HadoopRemoteAcceptor implements RemoteAcceptor {
String script = RemoteAcceptor.getScript(String.join(SPACE, args), this.shell);
if (this.useSugarPlugin)
script = SugarLoader.class.getPackage() + ".SugarLoader.load()\n" + script;
- final TraversalVertexProgram program = TraversalVertexProgram.build().traversal(this.hadoopGraph.getClass(), GraphTraversalSource.computer(), "gremlin-groovy", script).create();
+ final TraversalVertexProgram program = TraversalVertexProgram.build().traversal(this.hadoopGraph.getClass(), GraphTraversalSource.computer(), "gremlin-groovy", script).create(this.hadoopGraph);
final ComputerResult computerResult = this.hadoopGraph.compute().program(program).submit().get();
this.shell.getInterp().getContext().setProperty(RESULT, computerResult);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopCombine.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopCombine.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopCombine.java
index e7b94b5..c23d4dc 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopCombine.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopCombine.java
@@ -18,7 +18,9 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer;
+import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
@@ -45,7 +47,8 @@ public class HadoopCombine extends Reducer<ObjectWritable, ObjectWritable, Objec
@Override
public void setup(final Reducer<ObjectWritable, ObjectWritable, ObjectWritable, ObjectWritable>.Context context) {
- this.mapReduce = MapReduce.createMapReduce(ConfUtil.makeApacheConfiguration(context.getConfiguration()));
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(context.getConfiguration());
+ this.mapReduce = MapReduce.createMapReduce(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.mapReduce.workerStart(MapReduce.Stage.COMBINE);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopMap.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopMap.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopMap.java
index d88c134..3e55613 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopMap.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopMap.java
@@ -18,8 +18,10 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer;
+import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
@@ -44,7 +46,8 @@ public class HadoopMap extends Mapper<NullWritable, VertexWritable, ObjectWritab
@Override
public void setup(final Mapper<NullWritable, VertexWritable, ObjectWritable, ObjectWritable>.Context context) {
- this.mapReduce = MapReduce.createMapReduce(ConfUtil.makeApacheConfiguration(context.getConfiguration()));
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(context.getConfiguration());
+ this.mapReduce = MapReduce.createMapReduce(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.mapReduce.workerStart(MapReduce.Stage.MAP);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopReduce.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopReduce.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopReduce.java
index a0ddf67..66a14e3 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopReduce.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/HadoopReduce.java
@@ -18,7 +18,9 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer;
+import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
@@ -27,7 +29,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.Iterator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -44,7 +45,8 @@ public class HadoopReduce extends Reducer<ObjectWritable, ObjectWritable, Object
@Override
public void setup(final Reducer<ObjectWritable, ObjectWritable, ObjectWritable, ObjectWritable>.Context context) {
- this.mapReduce = MapReduce.createMapReduce(ConfUtil.makeApacheConfiguration(context.getConfiguration()));
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(context.getConfiguration());
+ this.mapReduce = MapReduce.createMapReduce(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.mapReduce.workerStart(MapReduce.Stage.REDUCE);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/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 d03fa37..dee7a29 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
@@ -211,6 +211,6 @@ public class GiraphGraphComputer extends AbstractHadoopGraphComputer implements
public static void main(final String[] args) throws Exception {
final FileConfiguration configuration = new PropertiesConfiguration(args[0]);
- new GiraphGraphComputer(HadoopGraph.open(configuration)).program(VertexProgram.createVertexProgram(configuration)).submit().get();
+ new GiraphGraphComputer(HadoopGraph.open(configuration)).program(VertexProgram.createVertexProgram(HadoopGraph.open(configuration),configuration)).submit().get();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
index eaac36f..ba5be5e 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMemory.java
@@ -18,13 +18,16 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
+import org.apache.commons.configuration.Configuration;
import org.apache.giraph.master.MasterCompute;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.MemoryHelper;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.io.DataInput;
@@ -64,7 +67,8 @@ public class GiraphMemory extends MasterCompute implements Memory {
public void compute() {
this.isMasterCompute = true;
if (0 == this.getSuperstep()) { // setup
- this.vertexProgram = VertexProgram.createVertexProgram(ConfUtil.makeApacheConfiguration(this.getConf()));
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(this.getConf());
+ this.vertexProgram = VertexProgram.createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.memoryKeys = new HashSet<>(this.vertexProgram.getMemoryComputeKeys());
try {
for (final String key : this.memoryKeys) {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMessageCombiner.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMessageCombiner.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMessageCombiner.java
index ef71207..7390c61 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMessageCombiner.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphMessageCombiner.java
@@ -18,9 +18,11 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
+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;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
@@ -49,7 +51,8 @@ public class GiraphMessageCombiner extends Combiner<ObjectWritable, ObjectWritab
@Override
public void setConf(final ImmutableClassesGiraphConfiguration configuration) {
this.configuration = configuration;
- this.messageCombiner = (MessageCombiner) VertexProgram.createVertexProgram(ConfUtil.makeApacheConfiguration(configuration)).getMessageCombiner().get();
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(configuration);
+ this.messageCombiner = (MessageCombiner) VertexProgram.createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration).getMessageCombiner().get();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
index 84a1888..0547aef 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/giraph/GiraphWorkerContext.java
@@ -18,9 +18,11 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.process.computer.giraph;
+import org.apache.commons.configuration.Configuration;
import org.apache.giraph.conf.GiraphConstants;
import org.apache.giraph.worker.WorkerContext;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
@@ -43,7 +45,8 @@ public final class GiraphWorkerContext extends WorkerContext {
}
public void preApplication() throws InstantiationException, IllegalAccessException {
- final VertexProgram vertexProgram = VertexProgram.createVertexProgram(ConfUtil.makeApacheConfiguration(this.getContext().getConfiguration()));
+ final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(this.getContext().getConfiguration());
+ final VertexProgram vertexProgram = VertexProgram.createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
this.vertexProgramPool = new VertexProgramPool(vertexProgram, this.getContext().getConfiguration().getInt(GiraphConstants.NUM_COMPUTE_THREADS.getKey(), 1));
this.memory = new GiraphMemory(this, vertexProgram);
this.deriveMemory = this.getContext().getConfiguration().getBoolean(Constants.GREMLIN_HADOOP_DERIVE_MEMORY, false);
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
index 5df9ac1..3cc0542 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/spark/SparkExecutor.java
@@ -32,6 +32,7 @@ import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.payload.Payloa
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.payload.ViewIncomingPayload;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.payload.ViewOutgoingPayload;
import org.apache.tinkerpop.gremlin.hadoop.process.computer.spark.payload.ViewPayload;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritableIterator;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
@@ -79,7 +80,7 @@ public final class SparkExecutor {
graphRDD.leftOuterJoin(viewIncomingRDD)) // every other iteration may have views and messages
// for each partition of vertices
.mapPartitionsToPair(partitionIterator -> {
- final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration); // each partition(Spark)/worker(TP3) has a local copy of the vertex program (a worker's task)
+ final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(HadoopGraph.open(apacheConfiguration),apacheConfiguration); // each partition(Spark)/worker(TP3) has a local copy of the vertex program (a worker's task)
final Set<String> elementComputeKeys = workerVertexProgram.getElementComputeKeys(); // the compute keys as a set
final String[] elementComputeKeysArray = elementComputeKeys.size() == 0 ? EMPTY_ARRAY : elementComputeKeys.toArray(new String[elementComputeKeys.size()]); // the compute keys as an array
final SparkMessenger<M> messenger = new SparkMessenger<>();
@@ -107,7 +108,7 @@ public final class SparkExecutor {
})).setName("viewOutgoingRDD");
// "message pass" by reducing on the vertex object id of the view and message payloads
- final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration).getMessageCombiner().orElse(null);
+ final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration).getMessageCombiner().orElse(null);
final JavaPairRDD<Object, ViewIncomingPayload<M>> newViewIncomingRDD = viewOutgoingRDD
.flatMapToPair(tuple -> () -> IteratorUtils.<Tuple2<Object, Payload>>concat(
IteratorUtils.of(new Tuple2<>(tuple._1(), tuple._2().getView())), // emit the view payload
@@ -161,7 +162,7 @@ public final class SparkExecutor {
public static <K, V> JavaPairRDD<K, V> executeMap(final JavaPairRDD<Object, VertexWritable> graphRDD, final MapReduce<K, V, ?, ?, ?> mapReduce, final Configuration apacheConfiguration) {
JavaPairRDD<K, V> mapRDD = graphRDD.mapPartitionsToPair(partitionIterator -> {
- final MapReduce<K, V, ?, ?, ?> workerMapReduce = MapReduce.<MapReduce<K, V, ?, ?, ?>>createMapReduce(apacheConfiguration);
+ final MapReduce<K, V, ?, ?, ?> workerMapReduce = MapReduce.<MapReduce<K, V, ?, ?, ?>>createMapReduce(HadoopGraph.open(apacheConfiguration),apacheConfiguration);
workerMapReduce.workerStart(MapReduce.Stage.MAP);
final SparkMapEmitter<K, V> mapEmitter = new SparkMapEmitter<>();
return () -> IteratorUtils.flatMap(partitionIterator, vertexWritable -> {
@@ -180,7 +181,7 @@ public final class SparkExecutor {
public static <K, V, OK, OV> JavaPairRDD<OK, OV> executeReduce(final JavaPairRDD<K, V> mapRDD, final MapReduce<K, V, OK, OV, ?> mapReduce, final Configuration apacheConfiguration) {
JavaPairRDD<OK, OV> reduceRDD = mapRDD.groupByKey().mapPartitionsToPair(partitionIterator -> {
- final MapReduce<K, V, OK, OV, ?> workerMapReduce = MapReduce.<MapReduce<K, V, OK, OV, ?>>createMapReduce(apacheConfiguration);
+ final MapReduce<K, V, OK, OV, ?> workerMapReduce = MapReduce.<MapReduce<K, V, OK, OV, ?>>createMapReduce(HadoopGraph.open(apacheConfiguration),apacheConfiguration);
workerMapReduce.workerStart(MapReduce.Stage.REDUCE);
final SparkReduceEmitter<OK, OV> reduceEmitter = new SparkReduceEmitter<>();
return () -> IteratorUtils.flatMap(partitionIterator, keyValue -> {
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/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 fff1039..133610f 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
@@ -182,6 +182,6 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
public static void main(final String[] args) throws Exception {
final FileConfiguration configuration = new PropertiesConfiguration(args[0]);
- new SparkGraphComputer(HadoopGraph.open(configuration)).program(VertexProgram.createVertexProgram(configuration)).submit().get();
+ new SparkGraphComputer(HadoopGraph.open(configuration)).program(VertexProgram.createVertexProgram(HadoopGraph.open(configuration),configuration)).submit().get();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
index 82fa2df..3afdc19 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/process/computer/util/MemoryMapReduce.java
@@ -25,6 +25,7 @@ import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.util.GraphComputerHelper;
import org.apache.tinkerpop.gremlin.process.computer.util.MapMemory;
import org.apache.tinkerpop.gremlin.process.computer.util.StaticMapReduce;
+import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.commons.configuration.Configuration;
@@ -62,7 +63,7 @@ public final class MemoryMapReduce extends StaticMapReduce<MapReduce.NullObject,
}
@Override
- public void loadState(final Configuration configuration) {
+ public void loadState(final Graph graph, final Configuration configuration) {
this.memoryKeys = new HashSet((List) configuration.getList(Constants.GREMLIN_HADOOP_MEMORY_KEYS));
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/48c15ee2/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableComparator.java
----------------------------------------------------------------------
diff --git a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableComparator.java b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableComparator.java
index 7ecdd7d..c810768 100644
--- a/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableComparator.java
+++ b/hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/ObjectWritableComparator.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.hadoop.structure.io;
+import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.util.ConfUtil;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.hadoop.conf.Configurable;
@@ -69,7 +70,8 @@ public abstract class ObjectWritableComparator implements RawComparator<ObjectWr
@Override
public void setConf(final Configuration configuration) {
this.configuration = configuration;
- this.comparator = MapReduce.<MapReduce<?,?,?,?,?>>createMapReduce(ConfUtil.makeApacheConfiguration(configuration)).getMapKeySort().get();
+ final org.apache.commons.configuration.Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(configuration);
+ this.comparator = MapReduce.<MapReduce<?,?,?,?,?>>createMapReduce(HadoopGraph.open(apacheConfiguration),apacheConfiguration).getMapKeySort().get();
}
}
@@ -77,7 +79,8 @@ public abstract class ObjectWritableComparator implements RawComparator<ObjectWr
@Override
public void setConf(final Configuration configuration) {
this.configuration = configuration;
- this.comparator = MapReduce.<MapReduce<?,?,?,?,?>>createMapReduce(ConfUtil.makeApacheConfiguration(configuration)).getReduceKeySort().get();
+ final org.apache.commons.configuration.Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(configuration);
+ this.comparator = MapReduce.<MapReduce<?,?,?,?,?>>createMapReduce(HadoopGraph.open(apacheConfiguration),apacheConfiguration).getReduceKeySort().get();
}
}
}