1. Overview

本文将通过闭包的概念引出累加器，并介绍累加器的基本使用以及如何自定义累加器，文章最后将说明在使用累加器时需要注意的问题

2. Basic

2.1 理解闭包

首先看一段代码

var counter = 0
var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach(x => counter += x)

println("Counter value: " + counter)

上述行为是未定义的( undefined )。这是因为 spark 作为一个分布式计算框架，在执行任务时，会将任务拆分成多个 task 分发给不同的 executor 来执行，而执行这些 task 前，spark 会计算 task 的闭包，上述代码中的闭包就是 foreach()。所谓闭包就是对 executors 可见的变量、方法或者代码片段。闭包会被序列化后发送给每个 executor。所以上述代码在 local 模式下，多个 executor 在一个 JVM 中时，使用的是同一个闭包，也就意味着 counter 使用的是同一个，那么更新操作可以同步到该 counter 上，结果是符合预期的。但如果是 cluster 模式，多个 executor 不在一个 JVM 执行，executor 可见的闭包是 driver 端序列化后发送过来的，各个 executor 更新的 counter 的值都是序列化后的副本值而不是 driver 端的 counter 本身，所以最终 counter 的值自然是 0。通过以上分析，可以得出在 spark 程序中需要注意两点: 1.闭包中引用的变量是可序列化的(否则无法发送到executor) 2.不要在闭包中修改闭包外的变量(行为未定义)。那是否意味着我们无法实现上述累加的功能了呢？spark 设计的时候显然考虑了这一常见需求，也就是累加器( Accumulators )

2.2 累加器的使用

var counterAccum = sc.longAccumulator("counter")
var rdd = sc.parallelize(1 to 10)

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

println("Counter value: " + counterAccum.value)

spark ui

通过 spark UI 可以看到最终正确算出了结果55，其中 4 个分区先各自计算自己的结果，最后在 driver 端聚合所有分区的结果。以上就是累加器的简单使用。

3. Deep

3.1 自定义累加器

3.1.1 累加器的原理

在学习自定义累加器前，如果搞清楚了累加器的原理，那么将事半功倍。这里通过一句话高度概括累加器的工作原理：首先序列化 driver 端 accumulator 到 executor ，序列化前调用 reset 重置 value 并使用 isZero 检测是否重置成功。单个 executor 内使用 add 进行累加，最终 driver 端对多个 executor 间的 accumulaotr 使用merge 进行合并得到结果。

3.1.2 自定义累加器

这里我们直接参考系统自带的 LongAccumulator 是如何实现的，毕竟源码的使用一定是最权威最规范的。

class LongAccumulator extends AccumulatorV2[jl.Long, jl.Long] {
  private var _sum = 0L
  private var _count = 0L
  def count: Long = _count
  def sum: Long = _sum
  def avg: Double = _sum.toDouble / _count
  override def value: jl.Long = _sum
  
  override def isZero: Boolean = _sum == 0L && _count == 0

  override def copy(): LongAccumulator = {
    val newAcc = new LongAccumulator
    newAcc._count = this._count
    newAcc._sum = this._sum
    newAcc
  }

  override def reset(): Unit = {
    _sum = 0L
    _count = 0L
  }

  override def add(v: jl.Long): Unit = {
    _sum += v
    _count += 1
  }

  def add(v: Long): Unit = {
    _sum += v
    _count += 1
  }

  override def merge(other: AccumulatorV2[jl.Long, jl.Long]): Unit = other match {
    case o: LongAccumulator =>
      _sum += o.sum
      _count += o.count
    case _ =>
      throw new UnsupportedOperationException(
        s"Cannot merge ${this.getClass.getName} with ${other.getClass.getName}")
  }

  private[spark] def setValue(newValue: Long): Unit = _sum = newValue

  
}

代码不难看懂，只要理解了累加器的工作原理，去实现对应的函数即可。这里有一点要注意的是，不同于系统自带的累加器，使用自定义累加器前需要注册：

// Then, create an Accumulator of this type:
val myVectorAcc = new VectorAccumulatorV2
// Then, register it into spark context:
sc.register(myVectorAcc, "MyVectorAcc1")

3.2 使用累加器中可能遇到的坑

当我们把累加器的操作放在 map 中执行的时候，后续如果有多个 action 操作共用该累加器的 RDD ，将会导致重复执行。也就意味着累加器会重复累加。为了避免这种错误，我们最好只在 action 算子如 foreach 中使用累加器，如果实在需要在 transformation 中使用，记得使用 cache 操作。