共享变量之累加器

Spark 累加器：Driver 的变量在 Executor 上改了没用？用累加器把数”收”回来

在 Spark 里，你可能会写出这样的代码：

var sum = 0
rdd.foreach(x => sum += x)
println(sum)   // 输出：0

直觉上应该是所有元素的和，但结果是 0。因为 sum 在 Driver 端定义，被复制到每个 Executor 上各自修改，改的是副本，Driver 端的 sum 纹丝不动。

累加器就是解决这个问题的：Driver 发一个”计数器”给 Executor，Executor 往里加数，Driver 最后把总数收回来。

累加器就干一件事：把 Executor 上的值聚合回 Driver。

累加器的基本用法

1. 创建累加器

val sumAcc = sc.longAccumulator("Sum")   // 长整型累加器
val countAcc = sc.doubleAccumulator("Count")  // 双精度累加器
val errAcc = sc.collectionAccumulator[String]("Errors")  // 集合累加器

2. Executor 上累加

rdd.foreach(x => sumAcc.add(x))   // 每个 Task 往累加器里加数

3. Driver 读取结果

println(sumAcc.value)   // 所有 Task 累加的总和

完整示例：

val rdd = sc.parallelize(1 to 100)
val sumAcc = sc.longAccumulator("Sum")

rdd.foreach(x => sumAcc.add(x))

println(sumAcc.value)   // 5050

累加器的类型

Spark 内置三种累加器：

类型	创建方法	用途
长整型	sc.longAccumulator("name")	计数、求和（int/long）
双精度	sc.doubleAccumulator("name")	求和（float/double）
集合	sc.collectionAccumulator[T]("name")	收集错误信息、特殊值

集合累加器最实用： 收集处理过程中遇到的异常数据。

val errorAcc = sc.collectionAccumulator[String]("Errors")

rdd.foreach { record =>
  if (record.contains("ERROR")) {
    errorAcc.add(record)
  }
}

errorAcc.value.forEach(println)   // 打印所有错误记录

自定义累加器

内置的不够用，自己写一个。继承 AccumulatorV2，实现 6 个方法：

import org.apache.spark.util.AccumulatorV2

// 字符串拼接累加器
class StringConcatAccumulator extends AccumulatorV2[String, String] {
  private var result: String = ""

  override def isZero: Boolean = result.isEmpty
  override def copy(): AccumulatorV2[String, String] = {
    val acc = new StringConcatAccumulator
    acc.result = this.result
    acc
  }
  override def reset(): Unit = result = ""
  override def add(v: String): Unit = result += v
  override def merge(other: AccumulatorV2[String, String]): Unit = {
    result += other.value
  }
  override def value: String = result
}

// 使用
val acc = new StringConcatAccumulator()
sc.register(acc, "StringConcat")
rdd.foreach(x => acc.add(x.toString + ","))
println(acc.value)   // "1,2,3,4,5,"

关键注意点：转换算子 vs 行动算子

累加器只在行动算子（Action）里可靠。 在转换算子（Transformation）里用累加器，可能因为惰性执行导致重复累加。

//  错误：在 map 里累加
val acc = sc.longAccumulator
val mapped = rdd.map(x => {
  acc.add(1)
  x * 2
})
mapped.count()   // 第一次：累加 100
mapped.count()   // 第二次：又累加 100  ← 重复了！
// 因为 map 被 count 触发了两次

//  正确：在 foreach 里累加
rdd.foreach(x => acc.add(1))  // 只执行一次

规则：累加器只放在 Action 里（foreach、collect、count），别放在 Transformation 里（map、filter）。

常见问题

1. 累加器值不对，比预期大

多半是累加器放在了 Transformation 里，被重复执行了。检查是不是在 map/filter 里用了累加器。

2. Executor 上想读累加器当前值，读不到

累加器是只写的，Executor 上只能 add，不能 value。想读值只能回到 Driver 读 acc.value。

3. 累加器更新慢

数据量大、分区多的时候，每个 Task 更新累加器都要序列化回 Driver。如果只是计数，可以考虑在每个分区先局部计数，最后汇总。