与 scala 并行创建地图的最佳方法是什么？

Question

假设我有一个应该转换为 Map 的集合，但不是像 map 方法那样以一对一的方式转换。

var map = collection.mutable.HashMap()
for (p <- dataList.par) {
  if(cond1(p)) {
    map += (p, true)
  } else {
    // do nothing
  }
}

我想出了几个解决方案，想知道什么是最好的。

1. map.synchronize { map += (p, true) }

2. 使用actor更新地图。但我不知道如何等到所有演员的任务完成

3. yield Some(p) or None 然后运行 ​​foreach { case Some(p) => map += (p, true)}。但是如果第一个迭代器来自并行集合，我不知道如何使其成为顺序。

Answer 1

不确定这实际上会表现最佳，但这应该使条件评估并行：

import scala.collection._
val map: mutable.Map[Int, Boolean] 
  = dataList.par.collect{case p if cond1(p) => (p, true)}(breakOut)

（使用可变 Map，因为它是您的代码所做的，但这不是必需的）。

上下文必须给出预期结果的类型（因此是: mutable.Map[Int, Boolean]），breakOut 才能工作。

编辑： breakOut 是 scala.collection.breakOut。返回集合的集合操作（​​此处为collect）采用隐式参数bf: CanBuildFrom[SourceCollectionType, ElementType, ResultType]。库提供的隐式CanBuildFroms 被安排以便返回最佳可能的 ResultType，并且最佳意味着最接近源集合类型。 breakOut 被传递来代替这个隐式参数，因此可以选择另一个CanBuildFrom，因此可以选择结果类型。 breakOut 所做的是选择 CanBuildFrom 而与源集合类型无关。但是有很多可用的隐含，并且没有优先级规则。这就是为什么结果类型必须由上下文给出的原因，以便可以选择隐式之一。

总而言之，当 breakOut 代替隐式参数传递时，结果将被构建为上下文中预期的类型。

Answer 2

另一个答案中提到的breakOut 解析为一个构建器工厂，用于收集map 的预期类型。 map 的预期类型是 mutable.Map[Int, Boolean]。

由于构建器工厂由顺序集合提供，collect 不会并行进行：

scala> val cond1: Int => Boolean = _ % 2 == 0
cond1: Int => Boolean = <function1>

scala> val dataList = 1 to 10
dataList: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala> val map: mutable.Map[Int,Boolean] = dataList.par.collect{case p if cond1(p) => println(Thread.currentThread); (p, true)}(breakOut)
Thread[Thread-8,5,main]
Thread[Thread-8,5,main]
Thread[Thread-8,5,main]
Thread[Thread-8,5,main]
Thread[Thread-8,5,main]
map: scala.collection.mutable.Map[Int,Boolean] = Map(10 -> true, 8 -> true, 4 -> true, 6 -> true, 2 -> true)

您可以从线程名称中看到 - 线程应该包含一个名称 ForkJoin-something。

正确的方法

正确的做法应该是首先使用breakOut，预期类型为并行映射，以便collect并行进行：

scala> val map: parallel.mutable.ParMap[Int,Boolean] = dataList.par.collect{case p if cond1(p) => println(Thread.currentThread);(p, true)}(breakOut)
Thread[Thread-9,5,main]
Thread[Thread-9,5,main]
Thread[Thread-9,5,main]
Thread[Thread-9,5,main]
Thread[Thread-9,5,main]
map: scala.collection.parallel.mutable.ParMap[Int,Boolean] = ParHashMap(10 -> true, 8 -> true, 4 -> true, 6 -> true, 2 -> true)

然后在collect 的结果上调用seq，因为seq 始终是O(1)。

更新：刚刚检查过 - 这似乎适用于主干，但不适用于 2.9.1.final。

补丁

但是，正如您所看到的，这也不起作用，因为它是一个错误，并将在下一个版本的 Scala 中修复。解决方法：

scala> val map: parallel.mutable.ParMap[Int, Boolean] = dataList.par.collect{case p if cond1(p) => println(Thread.currentThread);(p, true)}.map(x => x)(breakOut)
Thread[ForkJoinPool-1-worker-7,5,main]
Thread[ForkJoinPool-1-worker-3,5,main]
Thread[ForkJoinPool-1-worker-0,5,main]
Thread[ForkJoinPool-1-worker-8,5,main]
Thread[ForkJoinPool-1-worker-1,5,main]
map: scala.collection.parallel.mutable.ParMap[Int,Boolean] = ParHashMap(10 -> true, 8 -> true, 4 -> true, 6 -> true, 2 -> true)

scala> val sqmap = map.seq
sqmap: scala.collection.mutable.Map[Int,Boolean] = Map(10 -> true, 8 -> true, 4 -> true, 6 -> true, 2 -> true)

请注意，最终的map 目前将按顺序完成。

或者，如果您可以使用parallel.ParMap，您可以这样做：

scala> val map: Map[Int, Boolean] = dataList.par.collect{case p if cond1(p) => println(Thread.currentThread);(p, true)}.toMap.seq
Thread[ForkJoinPool-1-worker-2,5,main]
Thread[ForkJoinPool-1-worker-3,5,main]
Thread[ForkJoinPool-1-worker-7,5,main]
Thread[ForkJoinPool-1-worker-1,5,main]
Thread[ForkJoinPool-1-worker-8,5,main]
map: scala.collection.Map[Int,Boolean] = Map(10 -> true, 6 -> true, 2 -> true, 8 -> true, 4 -> true)