How to parallelize groupBy

Question

I was playing with .par and I'm wondering if the following computation can be further parallelized for performance gains or if there are other ways to compute the result faster. I don't think the end result depends on the order of the grouping so I'm hoping there are additional possible gains.

object Test {
  val data = (1 to 500000) map { i => (i % 100) -> (i % 10000) }

  def mutableIndex = {
    val map = collection.mutable.Map[Int, Set[Int]]().withDefaultValue(
      Set[Int]())
    for ((k, v) <- data) { map(k) = map(k) + v }
    map
  }

  def immutableIndex = data.groupBy(_._1).map{ case (k, seq) =>
    k -> seq.map(_._2).toSet
  }

  def immutableParIndex = data.par.groupBy(_._1).map{ case (k, seq) =>
    k -> seq.map(_._2).toSet
  }

  def main(args: Array[String]) {
    def bench(id: String)(block: => Unit) {
      val times = (new testing.Benchmark { def run() = block }).runBenchmark(10)
      println(id + " " + times + " sum: " + times.sum)
    }
    println("avail procs " + Runtime.getRuntime.availableProcessors)
    bench("mutable"){ mutableIndex }
    bench("immutable"){ immutableIndex }
    bench("immutable par"){ immutableParIndex }
  }

}

Running it prints this - using 2.9.1:

$ scalac -d classes -optimize A.scala
$ scala -cp classes Test
avail procs 4
mutable List(718, 343, 296, 297, 312, 312, 312, 312, 312, 312) sum: 3526
immutable List(312, 266, 266, 265, 265, 265, 265, 265, 249, 265) sum: 2683
immutable par List(546, 234, 234, 202, 187, 172, 188, 172, 187, 171) sum: 2293

Some notes:

• although the output above is pretty nice, the parallel version is also much more inconsistent depending on the constants I use in data and how many iteration I configure in bench (sometimes being less efficient than the sequential one). I wonder if it's expected of parallel collections.
• mutable gets faster as the set gets smaller (by decreasing the last modulo in data)
• if my benchmark is flawed, let me know how to fix it (e.g. I use the same data for all iterations, not sure if that skews the results)

---

Edit: here is a version based on concurrent hashmap and modeled after the library code for groupBy:

def syncIndex = {
  import collection.mutable.Builder
  import java.util.concurrent.ConcurrentHashMap
  import collection.JavaConverters._
  val m = new ConcurrentHashMap[Int, Builder[Int, Set[Int]]]().asScala
  for ((k, v) <- data.par) {
    val bldr = Set.newBuilder[Int]
    m.putIfAbsent(k, bldr) match {
      case Some(bldr) => bldr.synchronized(bldr += v)
      case None => bldr.synchronized(bldr += v)
    }
  }
  val b = Map.newBuilder[Int, Set[Int]]
  for ((k, v) <- m)
    b += ((k, v.result))
  b.result
}

It seeems to give a nice speed up on 2 cores but not on 4.

Answer 1

Not really an answer to your question, but I found .par gives a speedup especially on the Hotspot (32-bit?) Client, and not so much on Hotspot Server. I ran it in the REPL and the benchmark gets quicker on subsequent runs, since it's already warmed up.

I watched the processor usage on Task Manager and for each, and it goes from around 54% on the non-parallelized tasks to 75% on parallelized.

Java 7 also gives a pretty hefty speed boost.

Welcome to Scala version 2.9.0.1 (Java HotSpot(TM) Client VM, Java 1.6.0_22).

scala> Test.main(Array[String]())
avail procs 2
mutable List(1303, 1086, 1058, 1132, 1071, 1068, 1035, 1037, 1036, 1032) sum: 10858
immutable List(874, 872, 869, 856, 858, 857, 855, 855, 857, 849) sum: 8602
immutable par List(688, 502, 482, 479, 480, 465, 473, 473, 471, 472) sum: 4985

scala> Test.main(Array[String]())
avail procs 2
mutable List(1015, 1025, 1090, 1026, 1011, 1021, 1014, 1017, 1011, 1015) sum: 10245
immutable List(863, 868, 867, 865, 864, 883, 865, 863, 864, 864) sum: 8666
immutable par List(466, 468, 463, 466, 466, 469, 470, 467, 478, 467) sum: 4680

Welcome to Scala version 2.9.0.1 (Java HotSpot(TM) 64-Bit Server VM, Java 1.6.0_22).

scala> Test.main(Array[String]())
avail procs 2
mutable List(841, 360, 348, 338, 337, 338, 338, 342, 336, 336) sum: 3914
immutable List(320, 303, 302, 300, 304, 302, 305, 299, 305, 299) sum: 3039
immutable par List(521, 284, 244, 244, 232, 267, 209, 219, 231, 203) sum: 2654

scala> Test.main(Array[String]())
avail procs 2
mutable List(370, 393, 351, 342, 336, 343, 342, 340, 334, 340) sum: 3491
immutable List(301, 301, 302, 305, 300, 299, 303, 305, 304, 301) sum: 3021
immutable par List(207, 240, 201, 194, 204, 194, 197, 211, 207, 208) sum: 2063

scala> Test.main(Array[String]())
avail procs 2
mutable List(334, 336, 338, 339, 340, 338, 341, 334, 336, 340) sum: 3376
immutable List(300, 303, 297, 301, 298, 305, 302, 304, 296, 296) sum: 3002
immutable par List(194, 200, 190, 201, 192, 191, 195, 196, 202, 189) sum: 1950

Welcome to Scala version 2.9.0.1 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0).

scala> Test.main(Array[String]())
avail procs 2
mutable List(763, 258, 227, 235, 238, 279, 245, 227, 227, 243) sum: 2942
immutable List(274, 233, 228, 235, 238, 247, 243, 229, 233, 245) sum: 2405
immutable par List(635, 303, 261, 258, 217, 291, 204, 248, 219, 184) sum: 2820

scala> Test.main(Array[String]())
avail procs 2
mutable List(229, 229, 229, 230, 234, 226, 227, 227, 227, 232) sum: 2290
immutable List(228, 247, 231, 234, 210, 210, 209, 211, 210, 210) sum: 2200
immutable par List(173, 209, 160, 157, 158, 177, 179, 164, 163, 159) sum: 1699

scala> Test.main(Array[String]())
avail procs 2
mutable List(222, 218, 216, 214, 216, 215, 215, 219, 219, 218) sum: 2172
immutable List(211, 210, 211, 211, 212, 215, 215, 210, 211, 210) sum: 2116
immutable par List(161, 158, 168, 158, 156, 161, 150, 156, 163, 175) sum: 1606