You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2015/04/30 00:52:21 UTC
spark git commit: [SPARK-7181] [CORE] fix inifite loop in
Externalsorter's mergeWithAggregation
Repository: spark
Updated Branches:
refs/heads/master d7dbce8f7 -> 7f4b58373
[SPARK-7181] [CORE] fix inifite loop in Externalsorter's mergeWithAggregation
see [SPARK-7181](https://issues.apache.org/jira/browse/SPARK-7181).
Author: Qiping Li <li...@gmail.com>
Closes #5737 from chouqin/externalsorter and squashes the following commits:
2924b93 [Qiping Li] fix inifite loop in Externalsorter's mergeWithAggregation
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/7f4b5837
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7f4b5837
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7f4b5837
Branch: refs/heads/master
Commit: 7f4b583733714bbecb43fb0823134bf2ec720a17
Parents: d7dbce8
Author: Qiping Li <li...@gmail.com>
Authored: Wed Apr 29 23:52:16 2015 +0100
Committer: Sean Owen <so...@cloudera.com>
Committed: Wed Apr 29 23:52:16 2015 +0100
----------------------------------------------------------------------
.../scala/org/apache/spark/util/collection/ExternalSorter.scala | 3 ++-
.../org/apache/spark/util/collection/ExternalSorterSuite.scala | 5 ++++-
2 files changed, 6 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/7f4b5837/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
index ef3cac6..4ed8a74 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala
@@ -527,7 +527,8 @@ private[spark] class ExternalSorter[K, V, C](
val k = elem._1
var c = elem._2
while (sorted.hasNext && sorted.head._1 == k) {
- c = mergeCombiners(c, sorted.head._2)
+ val pair = sorted.next()
+ c = mergeCombiners(c, pair._2)
}
(k, c)
}
http://git-wip-us.apache.org/repos/asf/spark/blob/7f4b5837/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala b/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
index 9ff067f..de26aa3 100644
--- a/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSuite.scala
@@ -506,7 +506,10 @@ class ExternalSorterSuite extends FunSuite with LocalSparkContext with PrivateMe
val agg = new Aggregator[Int, Int, Int](i => i, (i, j) => i + j, (i, j) => i + j)
val ord = implicitly[Ordering[Int]]
val sorter = new ExternalSorter(Some(agg), Some(new HashPartitioner(3)), Some(ord), None)
- sorter.insertAll((0 until 100000).iterator.map(i => (i / 2, i)))
+
+ // avoid combine before spill
+ sorter.insertAll((0 until 50000).iterator.map(i => (i , 2 * i)))
+ sorter.insertAll((0 until 50000).iterator.map(i => (i, 2 * i + 1)))
val results = sorter.partitionedIterator.map{case (p, vs) => (p, vs.toSet)}.toSet
val expected = (0 until 3).map(p => {
(p, (0 until 50000).map(i => (i, i * 4 + 1)).filter(_._1 % 3 == p).toSet)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org