scala count word co-occurrence 性能真的很低

Question

当我尝试实现一个函数来计算 scala 中的单词共现时，我发现我的函数性能非常低。

共现这个词是：
也就是说我们有一个 List[List[Int]] （实际上是一个单词列表），
我们将为每个 List[Int] 生成一个组合，
然后我们将所有组合合并到一个映射中，并对每个重复键的值求和。

组合：
[0,1,2] -> [((0,1),1),((0,2),1),((1,2),1)]

合并组合：
[((0,1),1),((0,2),1),((1,2),1)] + [((0,1),1),((0,2), 1),((1,2),1)] =
HashMap{(0,1):2,(0,2):2,(1,2):2}

这是 scala 版本：

val arr = Array.range(0, 1000)
val counter = scala.collection.mutable.HashMap[(Int, Int), Int](  )
arr.combinations(2).toArray.map{
    row=>
        val key = (row(0), row(1))
        if (!counter.contains(key)) {
            counter(key) = 1
        }
        else {
            counter(key) += 1
        }
}
assert(counter.size == 499500)

Scala 版本 2：

val counter = arr.combinations(2).map(x => ((x(0),x(1)), 1)).toArray
.groupBy(_._1).mapValues(_.map(_._2).sum)

这里是python版本：

import itertools    
arr = range(0, 1000)
combs = list(itertools.combinations(arr, 2))
counter = dict()
for key in combs:
    try:
        counter[key] += 1
    except KeyError:
        counter[key] = 1
assert len(counter) == 499500

scala 版本都需要 9 秒，而 python 版本需要 1 秒。
我认为我的代码肯定有问题，但我想不出其他方法来改进它（我对 scala 很陌生）。

另外，我使用 mutable.HashMap 的原因是我想减少内存使用量。

任何帮助将不胜感激，谢谢。

Answer 1

问题在于集合中的 combine 方法。它创建了一个效率不高的迭代器。我已经创建了另一个示例，它在不使用组合的情况下快 10 倍：

  def time[R](block: => R): R = {
    val t0 = System.currentTimeMillis()
    val result = block    // call-by-name
    val t1 = System.currentTimeMillis()
    println("Elapsed time: " + (t1 - t0) + "ms")
    result
  }

  val arr = Array.range(0, 1000).toList

  def combinations2[A](input: List[A]): Iterator[(A, A)] =
    input.tails.flatMap(_ match {
      case h :: t => t.iterator.map((h, _))
      case Nil => Iterator.empty
    })

  val counter = scala.collection.mutable.HashMap[(Int, Int), Int](  )
  time {
    combinations2(arr).foreach {
      row =>
        val key = row
        if (!counter.contains(key)) {
          counter(key) = 1
        }
        else {
          counter(key) += 1
        }
    }
    assert(counter.size == 499500)
  }

看看这个

Answer 2

您需要将arr 转换为并行集合。理想情况下，到 RDD。因此，创建一个 spark 上下文，从您的数组中获取一个 RDD，如下所示，然后在其上运行您的操作。

val arr: RDD[Int] = sparkContext.parallelize(Array.range(0, 1000))

你真的应该查一下some tutorials