Scala的并行频率计算不起作用

Question

我学习Scala从书“斯卡拉不耐烦”工作的练习。请参阅以下问题以及我的答案和代码。我想知道我的答案是否正确。此外代码不起作用（所有频率都是1）。错误在哪里？

Q10：哈利黑客读取文件到字符串并希望使用 并行采集同时更新上线的 部分信件的频率。他使用以下代码：

val frequencies = new scala.collection.mutable.HashMap[Char, Int] 
for (c <- str.par) frequencies(c) = frequencies.getOrElse(c, 0) + 1 

为什么这是一个可怕的想法？他怎样才能真正平行计算 ？

我的回答： 这不是一个好主意，因为如果2个线程同时更新相同的频率，结果是不确定的。

我的代码：

def parFrequency(str: String) = { 
    str.par.aggregate(Map[Char, Int]())((m, c) => { m + (c -> (m.getOrElse(c, 0) + 1)) }, _ ++ _) 
} 

单元测试：

"Method parFrequency" should "return the frequency of each character in a string" in { 
    val freq = parFrequency("harry hacker") 

    freq should have size 8 

    freq('h') should be(2) // fails 
    freq('a') should be(2) 
    freq('r') should be(3) 
    freq('y') should be(1) 
    freq(' ') should be(1) 
    freq('c') should be(1) 
    freq('k') should be(1) 
    freq('e') should be(1) 
} 

编辑： 阅读this线后，我更新的代码。现在，如果单独运行测试，但如果作为套件运行则失败。

def parFrequency(str: String) = { 
    val freq = ImmutableHashMap[Char, Int]() 
    str.par.aggregate(freq)((_, c) => ImmutableHashMap(c -> 1), (m1, m2) => m1.merged(m2)({ 
     case ((k, v1), (_, v2)) => (k, v1 + v2) 
    })) 
} 

编辑2： 见下面我的解决方案。

Answer 1

这似乎工作。我喜欢它比这里提出的其他解决方案更好，因为：

1. 这是很多比implicit class和略少的代码比使用getOrElse与foldLeft代码更少。
2. 它使用API​​中的merged函数来打算做我想要的。

3. 这是我自己的解决方案:)

   def parFrequency(str: String) = { 
       val freq = ImmutableHashMap[Char, Int]() 
       str.par.aggregate(freq)((_, c) => ImmutableHashMap(c -> 1), _.merged(_) { 
       case ((k, v1), (_, v2)) => (k, v1 + v2) 
       }) 
   } 
   

感谢您抽出宝贵时间来帮助我。

Answer 2

++不结合相同的密钥的值。所以当你合并地图时，你会得到（对于共享键）其中一个值（在这种情况下总是1），而不是值的总和。

这工作：

def parFrequency(str: String) = { 
    str.par.aggregate(Map[Char, Int]())((m, c) => { m + (c -> (m.getOrElse(c, 0) + 1)) }, 
    (a,b) => b.foldLeft(a){case (acc, (k,v))=> acc updated (k, acc.getOrElse(k,0) + v) }) 
} 

val freq = parFrequency("harry hacker") 
//> Map(e -> 1, y -> 1, a -> 2, -> 1, c -> 1, h -> 2, r -> 3, k -> 1) 

的foldLeft迭代的地图一，更新其它地图的键/值发现。

Answer 3

你麻烦第一种情况下，你自己检测是++运营商刚刚串联，下降相同的密钥的第二occurence。

现在在第二种情况下，你有(_, c) => ImmutableHashMap(c -> 1)，它只是删除我在seqop阶段发现的所有字符。

我的建议是延长Map类型有特殊compination操作，在HashMap像merged工作，并保持在seqop阶段从第一例的集：

implicit class MapUnionOps[K, V](m1: Map[K, V]) { 
    def unionWith[V1 >: V](m2: Map[K, V1])(f: (V1, V1) => V1): Map[K, V1] = { 
    val kv1 = m1.filterKeys(!m2.contains(_)) 
    val kv2 = m2.filterKeys(!m1.contains(_)) 
    val common = (m1.keySet & m2.keySet).toSeq map (k => (k, f(m1(k), m2(k)))) 
    (common ++ kv1 ++ kv2).toMap 
    } 
} 

def parFrequency(str: String) = { 
    str.par.aggregate(Map[Char, Int]())((m, c) => {m + (c -> (m.getOrElse(c, 0) + 1))}, (m1, m2) => (m1 unionWith m2)(_ + _)) 
} 

或者你可以使用从保罗的回答fold解决方案，但为更好的表现为每个合并选择较小的地图遍历：

implicit class MapUnionOps[K, V](m1: Map[K, V]) { 
    def unionWith(m2: Map[K, V])(f: (V, V) => V): Map[K, V] = 
    if (m2.size > m1.size) m2.unionWith(m1)(f) 
    else m2.foldLeft(m1) { 
     case (acc, (k, v)) => acc + (k -> acc.get(k).fold(v)(f(v, _))) 
    } 
}