自定义累加器实现wordCount的效果代码

自定义累加器在Spark中用于在多个action算子触发的job中重复累加数据，并且需要action算子才能触发累加操作5。以下是实现WordCount效果的自定义累加器的示例代码：

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.SparkContext

object WordCountAccumulator {
  // 定义一个累加器，用于WordCount
  class WordCountAccumulator extends AccumulatorV2[String, Int] {
    // 用于存储单词计数的变量
    private var count = Map[String, Int]()

    // 只用于累加器内部的变量，用于更新count
    private def newValues: mutable.Map[String, Int] = {
      val newCount = count.clone()
      newCount
    }

    // 用于更新累加器的值
    def add(word: String): Unit = {
      val newCount = newValues
      newCount(word) = newCount.getOrElse(word, 0) + 1
      count = newCount
    }

    // 用于合并两个累加器的值
    def merge(other: AccumulatorV2[String, Int]): Unit = {
      val otherCount = other.value
      count = count ++ otherCount.map { case (word, cnt) => word -> (count.getOrElse(word, 0) + cnt) }
    }

    // 返回累加器的值
    def value: Int = count.values.sum

    // 用于在Driver端获取最终的单词计数
    def toLocal: Map[String, Int] = count
  }

  // 创建累加器的辅助方法
  def create(): WordCountAccumulator = new WordCountAccumulator
}

// 使用自定义累加器进行WordCount
val conf = new SparkConf().setAppName("CustomWordCount")
val sc = new SparkContext(conf)
val accumulator = WordCountAccumulator.create()

val input = sc.parallelize(List("hello world", "hello scala", "akka spark"))
input.flatMap(_.split(" ")).foreach(word => accumulator.add(word))

println(accumulator.value) // 打印总单词数
sc.stop()

在上述代码中，WordCountAccumulator 类继承自 AccumulatorV2，实现了 add, merge, value, 和 toLocal 方法。add 方法用于更新单词计数，merge 方法用于合并累加器，value 方法返回总单词数，而 toLocal 方法返回单词计数的本地Map表示34789。