Spark：无法使用scala在元组/计数上使用累加器

Question

我正在尝试用累加器逻辑替换 reduceByKey 以进行字数统计。

wc.txt

你好，你好吗

这是我目前得到的：

val words = sc.textFile("wc.txt").flatMap(_.split(" "))
val accum = sc.accumulator(0,"myacc")
for (i <- 1 to words.count.toInt) 
    foreach( x => accum+ =x)
    .....

如何处理它。任何想法或想法表示赞赏。

Answer 1

确实，为此使用累加器很麻烦且不推荐 - 但为了完整起见 - 以下是它的完成方式（至少对于 Spark 版本 1.6 ）。请注意，这使用了一个已弃用的 API，该 API 不会成为下一个版本的一部分。

您需要一个Map[String, Long] 累加器，默认情况下不可用，因此您需要创建自己的AccumulableParam 实现并隐式使用它：

// some data:
val words = sc.parallelize(Seq("Hello how are are you")).flatMap(_.split(" "))

// aliasing the type, just for convenience
type AggMap = Map[String, Long]

// creating an implicit AccumulableParam that counts by String key
implicit val param: AccumulableParam[AggMap, String] = new AccumulableParam[AggMap, String] {
  // increase matching value by 1, or create it if missing
  override def addAccumulator(r: AggMap, t: String): AggMap = 
    r.updated(t, r.getOrElse(t, 0L) + 1L)

  // merge two maps by summing matching values 
  override def addInPlace(r1: AggMap, r2: AggMap): AggMap = 
    r1 ++ r2.map { case (k, v) => k -> (v + r1.getOrElse(k, 0L)) }

  // start with an empty map
  override def zero(initialValue: AggMap): AggMap = Map.empty
}

// create the accumulator; This will use the above `param` implicitly
val acc = sc.accumulable[AggMap, String](Map.empty[String, Long])

// add each word to accumulator; the `count()` can be replaced by any Spark action - 
// we just need to trigger the calculation of the mapped RDD 
words.map(w => { acc.add(w); w }).count()

// after the action, we acn read the value of the accumulator
val result: AggMap = acc.value

result.foreach(println)
// (Hello,1)
// (how,1)
// (are,2)
// (you,1)

Answer 2

据我了解，您想使用 Spark 累加器计算文本文件中的所有单词，在这种情况下，您可以使用：

words.foreach(_ => accum.add(1L))