scala - 如何并行化groupBy

Question

我正在使用.par，我想知道是否可以进一步并行执行以下计算以提高性能，或者是否还有其他方法可以更快地计算结果。我不认为最终结果取决于分组的顺序，所以我希望还有其他可能的 yield 。

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 }
  }

}

运行它会打印此内容-使用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

一些注意事项:

尽管上面的输出非常好，但是并行版本也更加不一致，具体取决于我在data中使用的常量以及在bench中配置的迭代次数(有时效率不如顺序代码)。我想知道是否期望并行收集。

随着集合变小，

mutable变快(通过减少数据的最后模数)

(如果我的基准测试有缺陷)，请让我知道如何解决(例如，我对所有迭代使用相同的数据，不确定是否会使结果产生偏差)

编辑:这是一个基于并发哈希图的版本，并根据 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
}

它似乎可以在2个内核上提供不错的速度，但在4个内核上却不能。

Answer 1

并不是您问题的真正答案，但是我发现.par尤其在Hotspot(32位？)客户端上提供了加速，而在Hotspot Server上却没有那么快。我已经在REPL中运行了它，并且由于已经预热了基准，因此以后的运行会更快。

我在任务管理器和每个任务管理器上都观察到处理器的使用情况，它从非并行任务的约54％变为并行任务的75％。

Java 7还大大提高了速度。

欢迎使用Scala 2.9.0.1版(Java HotSpot(TM)客户端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

欢迎使用Scala 2.9.0.1版(Java HotSpot™64位服务器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

欢迎使用Scala 2.9.0.1版(Java HotSpot™64位服务器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