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/12/09 01:25:08 UTC
incubator-tinkerpop git commit: provided a general way to ensure that
the number of parititions is NOT larger than the number of workers. Moreover
RDD.coalesce() does not require a shuffle as it only handle partition
reduction.
Repository: incubator-tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1023 10d9b16d3 -> aef952832
provided a general way to ensure that the number of parititions is NOT larger than the number of workers. Moreover RDD.coalesce() does not require a shuffle as it only handle partition reduction.
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/aef95283
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/aef95283
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/aef95283
Branch: refs/heads/TINKERPOP-1023
Commit: aef95283207b8b22a236151ba6084fd193c73401
Parents: 10d9b16
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Tue Dec 8 17:24:57 2015 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Tue Dec 8 17:24:57 2015 -0700
----------------------------------------------------------------------
.../gremlin/spark/process/computer/SparkGraphComputer.java | 8 +++++---
.../spark/process/computer/SparkHadoopGraphProvider.java | 4 ----
2 files changed, 5 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/aef95283/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 c9b57b8..9219211 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
@@ -142,7 +142,7 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
hadoopConfiguration.forEach(entry -> sparkConfiguration.set(entry.getKey(), entry.getValue()));
// execute the vertex program and map reducers and if there is a failure, auto-close the spark context
try {
- final JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
+ final JavaSparkContext sparkContext = new JavaSparkContext(SparkContext.getOrCreate(sparkConfiguration));
Spark.create(sparkContext.sc()); // this is the context RDD holder that prevents GC
updateLocalConfiguration(sparkContext, sparkConfiguration);
// add the project jars to the cluster
@@ -152,8 +152,10 @@ public final class SparkGraphComputer extends AbstractHadoopGraphComputer {
try {
graphRDD = hadoopConfiguration.getClass(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, InputFormatRDD.class, InputRDD.class)
.newInstance()
- .readGraphRDD(apacheConfiguration, sparkContext)
- .cache();
+ .readGraphRDD(apacheConfiguration, sparkContext);
+ if (graphRDD.partitions().size() > this.workers) // ensures that the graphRDD does not have more partitions than workers
+ graphRDD = graphRDD.coalesce(this.workers);
+ graphRDD = graphRDD.cache();
} catch (final InstantiationException | IllegalAccessException e) {
throw new IllegalStateException(e.getMessage(), e);
}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/aef95283/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
----------------------------------------------------------------------
diff --git a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
index 878b38a..f14cf7e 100644
--- a/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
+++ b/spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkHadoopGraphProvider.java
@@ -22,10 +22,8 @@ import org.apache.tinkerpop.gremlin.GraphProvider;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.HadoopGraphProvider;
-import org.apache.tinkerpop.gremlin.process.computer.bulkloading.BulkLoaderVertexProgramTest;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.engine.ComputerTraversalEngine;
-import org.apache.tinkerpop.gremlin.spark.structure.Spark;
import org.apache.tinkerpop.gremlin.spark.structure.io.ToyGraphInputRDD;
import org.apache.tinkerpop.gremlin.spark.structure.io.gryo.GryoSerializer;
import org.apache.tinkerpop.gremlin.structure.Graph;
@@ -46,8 +44,6 @@ public final class SparkHadoopGraphProvider extends HadoopGraphProvider {
final Map<String, Object> config = super.getBaseConfiguration(graphName, test, testMethodName, loadGraphWith);
config.put(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true); // this makes the test suite go really fast
if (null != loadGraphWith && RANDOM.nextBoolean()) {
- if (test.equals(BulkLoaderVertexProgramTest.class))
- Spark.close(); // this forces workers to be 1 which is required for BulkLoaderVertexProgramTest
config.put(Constants.GREMLIN_SPARK_GRAPH_INPUT_RDD, ToyGraphInputRDD.class.getCanonicalName());
// config.put(Constants.GREMLIN_HADOOP_GRAPH_INPUT_FORMAT, InputRDDFormat.class.getCanonicalName());
}