使用`aggregate`合并地图

Question

例如,对于任何给定的集合Map,

val in = Array( Map("a" -> 1,  "b" -> 2),
                Map("a" -> 11, "c" -> 4),
                Map("b" -> 7,  "c" -> 10))

如何使用aggregate上in.par,从而地图合并成

Map ( "a" -> 12, "b" -> 9, "c" -> 14 )

注意Map合并已被多次询问,但寻找aggregate并行集合的解决方案.

非常感谢

Answer 1

seqop将合并应用为and怎么样comboop？

val in = Array(
  Map("a" -> 1,  "b" -> 2),
  Map("a" -> 11, "c" -> 4),
  Map("b" -> 7,  "c" -> 10)
)

def merge(m1: Map[String, Int], m2: Map[String, Int]): Map[String, Int] =
  m1 ++ m2.map { case (k, v) => k -> (v + m1.getOrElse(k, 0)) }

in.par.aggregate(Map[String, Int]())(merge, merge)

更新

您传递给aggregate初始累加器值（空映射）和两个闭包 -seqop和comboop。

并行序列分为多个分区以并行处理。通过连续应用于累加器和数组元素来处理每个分区seqop。

def seqop(
    accumulator: Map[String, Int], 
    element: Map[String, Int]): Map[String, Int] = merge(accumulator, element)

seqop获取初始累加器值和第一个数组元素并将其合并。接下来，它将获取上一个结果和下一个数组元素，依此类推，直到整个分区合并到一个映射中。

当每个分区合并到一个单独的映射中时，应通过应用来组合这些映射comboop。comboop从第一个分区获取合并的映射，从第二个分区获取合并的映射，并将其合并在一起。接下来，它从第三个分区获取先前的结果和映射，依此类推，直到所有内容合并到一个映射中。这是 的结果aggregate。

def comboop(
    m1: Map[String, Int], 
    m2: Map[String, Int]): Map[String, Int] = merge(m1, m2)

seqop和comboop是相同的只是巧合。一般来说，它们在逻辑和签名上有所不同。