diff options
-rw-r--r-- | core/src/main/scala/spark/rdd/CoalescedRDD.scala | 2 | ||||
-rw-r--r-- | core/src/test/scala/spark/RDDSuite.scala | 10 |
2 files changed, 6 insertions, 6 deletions
diff --git a/core/src/main/scala/spark/rdd/CoalescedRDD.scala b/core/src/main/scala/spark/rdd/CoalescedRDD.scala index f46dd1ee6c..205175edfc 100644 --- a/core/src/main/scala/spark/rdd/CoalescedRDD.scala +++ b/core/src/main/scala/spark/rdd/CoalescedRDD.scala @@ -310,7 +310,7 @@ class PartitionCoalescer(maxPartitions: Int, prev: RDD[_], balanceSlack: Double) for ((p,i) <- prev.partitions.zipWithIndex) { groupArr(i).list += p } - } else { // old code, just splits them grouping partitions that are next to each other + } else { // no locality available, then simply split partitions based on positions in array (0 until maxPartitions).foreach { i => val rangeStart = ((i.toLong * prev.partitions.length) / maxPartitions).toInt val rangeEnd = (((i.toLong + 1) * prev.partitions.length) / maxPartitions).toInt diff --git a/core/src/test/scala/spark/RDDSuite.scala b/core/src/test/scala/spark/RDDSuite.scala index a757aebd65..8b6fa9c81e 100644 --- a/core/src/test/scala/spark/RDDSuite.scala +++ b/core/src/test/scala/spark/RDDSuite.scala @@ -184,7 +184,7 @@ class RDDSuite extends FunSuite with SharedSparkContext { assert(splits.length === 3) // ensure it indeed created 3 partitions assert(splits.foldLeft(true) - ( (x,y) => if (!x) false else y.length >= 2) === true) // (2+ balance) + ( (x,y) => if (!x) false else y.length >= 1) === true) // (1+ balance) // If we try to coalesce into more partitions than the original RDD, it should just // keep the original number of partitions. @@ -208,17 +208,17 @@ class RDDSuite extends FunSuite with SharedSparkContext { val data2 = sc.makeRDD(blocks) val coalesced2 = data2.coalesce(numMachines*2) - // test that you get over 95% locality in each group + // test that you get over 90% locality in each group val minLocality = coalesced2.partitions .map( part => part.asInstanceOf[CoalescedRDDPartition].localFraction ) .foldLeft(100.)( (perc, loc) => math.min(perc,loc) ) - assert(minLocality > 0.95) + assert(minLocality > 0.90) - // test that the groups are load balanced with 100 +/- 15 elements in each + // test that the groups are load balanced with 100 +/- 20 elements in each val maxImbalance = coalesced2.partitions .map( part => part.asInstanceOf[CoalescedRDDPartition].parents.size ) .foldLeft(0)((dev, curr) => math.max(math.abs(100-curr),dev)) - assert(maxImbalance < 15) + assert(maxImbalance < 20) } test("zipped RDDs") { |