git commit: [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
Repository: spark
Updated Branches:
refs/heads/master 1bed0a386 -> 00362dac9
[HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
9b225ac3072de522b40b46aba6df1f1c231f13ef has been causing GraphX tests
to fail nondeterministically, which is blocking development for others.
Author: Ankur Dave
Closes #2271 from ankurdave/SPARK-3400 and squashes the following commits:
10c2a97 [Ankur Dave] [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD
zipPartitions"
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/00362dac
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/00362dac
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/00362dac
Branch: refs/heads/master
Commit: 00362dac976cd05b06638deb11d990d612429e0b
Parents: 1bed0a3
Author: Ankur Dave
Authored: Wed Sep 3 23:49:47 2014 -0700
Committer: Ankur Dave
Committed: Wed Sep 3 23:49:47 2014 -0700
--
.../scala/org/apache/spark/graphx/EdgeRDD.scala | 4 ++--
.../scala/org/apache/spark/graphx/GraphSuite.scala | 16
2 files changed, 2 insertions(+), 18 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/00362dac/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
--
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
index 35fbd47..5bcb96b 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
@@ -19,7 +19,7 @@ package org.apache.spark.graphx
import scala.reflect.{classTag, ClassTag}
-import org.apache.spark._
+import org.apache.spark.{OneToOneDependency, Partition, Partitioner,
TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
@@ -55,7 +55,7 @@ class EdgeRDD[@specialized ED: ClassTag, VD: ClassTag](
* partitioner that allows co-partitioning with `partitionsRDD`.
*/
override val partitioner =
-partitionsRDD.partitioner.orElse(Some(new
HashPartitioner(partitionsRDD.partitions.size)))
+
partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD)))
override def compute(part: Partition, context: TaskContext):
Iterator[Edge[ED]] = {
val p = firstParent[(PartitionID, EdgePartition[ED, VD])].iterator(part,
context)
http://git-wip-us.apache.org/repos/asf/spark/blob/00362dac/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
--
diff --git a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
index eaaa449..6506bac 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
@@ -19,7 +19,6 @@ package org.apache.spark.graphx
import org.scalatest.FunSuite
-import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.Graph._
import org.apache.spark.graphx.PartitionStrategy._
@@ -351,19 +350,4 @@ class GraphSuite extends FunSuite with LocalSparkContext {
}
}
- test("non-default number of edge partitions") {
-val n = 10
-val defaultParallelism = 3
-val numEdgePartitions = 4
-assert(defaultParallelism != numEdgePartitions)
-val conf = new SparkConf()
- .set("spark.default.parallelism", defaultParallelism.toString)
-val sc = new SparkContext("local", "test", conf)
-val edges = sc.parallelize((1 to n).map(x => (x: VertexId, 0: VertexId)),
- numEdgePartitions)
-val graph = Graph.fromEdgeTuples(edges, 1)
-val neighborAttrSums = graph.mapReduceTriplets[Int](
- et => Iterator((et.dstId, et.srcAttr)), _ + _)
-assert(neighborAttrSums.collect.toSet === Set((0: VertexId, n)))
- }
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
git commit: [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
Repository: spark
Updated Branches:
refs/heads/branch-1.0 8dd7690e2 -> 4d3ab2925
[HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
9b225ac3072de522b40b46aba6df1f1c231f13ef has been causing GraphX tests
to fail nondeterministically, which is blocking development for others.
Author: Ankur Dave
Closes #2271 from ankurdave/SPARK-3400 and squashes the following commits:
10c2a97 [Ankur Dave] [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD
zipPartitions"
(cherry picked from commit 00362dac976cd05b06638deb11d990d612429e0b)
Signed-off-by: Ankur Dave
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/4d3ab292
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/4d3ab292
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/4d3ab292
Branch: refs/heads/branch-1.0
Commit: 4d3ab292576396e11d9e5c2dabb676c07b34d286
Parents: 8dd7690
Author: Ankur Dave
Authored: Wed Sep 3 23:49:47 2014 -0700
Committer: Ankur Dave
Committed: Wed Sep 3 23:50:22 2014 -0700
--
.../scala/org/apache/spark/graphx/EdgeRDD.scala | 4 ++--
.../scala/org/apache/spark/graphx/GraphSuite.scala | 16
2 files changed, 2 insertions(+), 18 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/4d3ab292/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
--
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
index 4dd15bf..a8fc095 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
@@ -19,7 +19,7 @@ package org.apache.spark.graphx
import scala.reflect.{classTag, ClassTag}
-import org.apache.spark._
+import org.apache.spark.{OneToOneDependency, Partition, Partitioner,
TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
@@ -45,7 +45,7 @@ class EdgeRDD[@specialized ED: ClassTag, VD: ClassTag](
* partitioner that allows co-partitioning with `partitionsRDD`.
*/
override val partitioner =
-partitionsRDD.partitioner.orElse(Some(new
HashPartitioner(partitionsRDD.partitions.size)))
+
partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD)))
override def compute(part: Partition, context: TaskContext):
Iterator[Edge[ED]] = {
val p = firstParent[(PartitionID, EdgePartition[ED, VD])].iterator(part,
context)
http://git-wip-us.apache.org/repos/asf/spark/blob/4d3ab292/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
--
diff --git a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
index e1b83c2..abc25d0 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
@@ -19,7 +19,6 @@ package org.apache.spark.graphx
import org.scalatest.FunSuite
-import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.Graph._
import org.apache.spark.graphx.PartitionStrategy._
@@ -326,19 +325,4 @@ class GraphSuite extends FunSuite with LocalSparkContext {
}
}
- test("non-default number of edge partitions") {
-val n = 10
-val defaultParallelism = 3
-val numEdgePartitions = 4
-assert(defaultParallelism != numEdgePartitions)
-val conf = new SparkConf()
- .set("spark.default.parallelism", defaultParallelism.toString)
-val sc = new SparkContext("local", "test", conf)
-val edges = sc.parallelize((1 to n).map(x => (x: VertexId, 0: VertexId)),
- numEdgePartitions)
-val graph = Graph.fromEdgeTuples(edges, 1)
-val neighborAttrSums = graph.mapReduceTriplets[Int](
- et => Iterator((et.dstId, et.srcAttr)), _ + _)
-assert(neighborAttrSums.collect.toSet === Set((0: VertexId, n)))
- }
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
git commit: [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
Repository: spark
Updated Branches:
refs/heads/branch-1.1 f41c45a75 -> 8c40ab5c0
[HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"
9b225ac3072de522b40b46aba6df1f1c231f13ef has been causing GraphX tests
to fail nondeterministically, which is blocking development for others.
Author: Ankur Dave
Closes #2271 from ankurdave/SPARK-3400 and squashes the following commits:
10c2a97 [Ankur Dave] [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD
zipPartitions"
(cherry picked from commit 00362dac976cd05b06638deb11d990d612429e0b)
Signed-off-by: Ankur Dave
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/8c40ab5c
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/8c40ab5c
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/8c40ab5c
Branch: refs/heads/branch-1.1
Commit: 8c40ab5c06ab72e85a8a9d4272fed0e81eca1d3a
Parents: f41c45a
Author: Ankur Dave
Authored: Wed Sep 3 23:49:47 2014 -0700
Committer: Ankur Dave
Committed: Wed Sep 3 23:50:11 2014 -0700
--
.../scala/org/apache/spark/graphx/EdgeRDD.scala | 4 ++--
.../scala/org/apache/spark/graphx/GraphSuite.scala | 16
2 files changed, 2 insertions(+), 18 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/8c40ab5c/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
--
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
index 0f1a101..899a3cb 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/EdgeRDD.scala
@@ -19,7 +19,7 @@ package org.apache.spark.graphx
import scala.reflect.{classTag, ClassTag}
-import org.apache.spark._
+import org.apache.spark.{OneToOneDependency, Partition, Partitioner,
TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
@@ -47,7 +47,7 @@ class EdgeRDD[@specialized ED: ClassTag, VD: ClassTag](
* partitioner that allows co-partitioning with `partitionsRDD`.
*/
override val partitioner =
-partitionsRDD.partitioner.orElse(Some(new
HashPartitioner(partitionsRDD.partitions.size)))
+
partitionsRDD.partitioner.orElse(Some(Partitioner.defaultPartitioner(partitionsRDD)))
override def compute(part: Partition, context: TaskContext):
Iterator[Edge[ED]] = {
val p = firstParent[(PartitionID, EdgePartition[ED, VD])].iterator(part,
context)
http://git-wip-us.apache.org/repos/asf/spark/blob/8c40ab5c/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
--
diff --git a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
index eaaa449..6506bac 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
@@ -19,7 +19,6 @@ package org.apache.spark.graphx
import org.scalatest.FunSuite
-import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.graphx.Graph._
import org.apache.spark.graphx.PartitionStrategy._
@@ -351,19 +350,4 @@ class GraphSuite extends FunSuite with LocalSparkContext {
}
}
- test("non-default number of edge partitions") {
-val n = 10
-val defaultParallelism = 3
-val numEdgePartitions = 4
-assert(defaultParallelism != numEdgePartitions)
-val conf = new SparkConf()
- .set("spark.default.parallelism", defaultParallelism.toString)
-val sc = new SparkContext("local", "test", conf)
-val edges = sc.parallelize((1 to n).map(x => (x: VertexId, 0: VertexId)),
- numEdgePartitions)
-val graph = Graph.fromEdgeTuples(edges, 1)
-val neighborAttrSums = graph.mapReduceTriplets[Int](
- et => Iterator((et.dstId, et.srcAttr)), _ + _)
-assert(neighborAttrSums.collect.toSet === Set((0: VertexId, n)))
- }
}
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
