Spark累加器返回值为预期两倍的问题排查

你在父类中声明Spark累加器，父类func1将DataFrame映射到子类实现的func2，子类func2中对累加器执行加1操作。预期映射函数被调用x次时累加器值为x，但实际结果约为2x。相关代码如下：

abstract class A{
  var accum = sparkCtx.accumulator(0)
  def func1(df) {
    var df2 = df.map(func2)
    func3(df2)
  }
  def func3(df2) {
    // 过滤条件永远不满足，结果为空DataFrame
    var df3 = df2.filter("Condition which never satisfies")
    var df4 = df3.map(func2)
}
class B extends A{
  def func2() {
    accum.add(1)
   // 返回数据
  }
}

错误原因 #

• Spark惰性求值导致重复计算：Spark的Transformation操作（如map、filter）是延迟执行的，只有遇到Action操作时才会触发整个血缘链的计算。你的代码中，df4依赖df3，而df3又依赖df2。当后续有Action触发df4的计算时，Spark会回溯计算df2；如果df2没有被持久化，Spark会重新执行df.map(func2)，导致func2被调用两次——一次生成df2，一次为了确认df3的过滤结果（即便df3最终是空的）。
• Transformation中使用累加器的风险：累加器设计为在Action操作中使用时保证最终结果正确，但在Transformation中使用时，由于Transformation可能被Spark多次重算（比如节点失败重试、血缘链重新计算），会导致累加器被重复更新。

解决办法 #

• 避免在Transformation中更新累加器：将累加器的更新逻辑移到Action操作中，比如使用foreach替代map来更新累加器，因为Action只会触发一次计算。
• 持久化中间DataFrame：在生成df2后调用df2.cache()或df2.persist()，将中间结果缓存到内存或磁盘，避免后续计算时重新执行上游的map(func2)。修改后的func1示例：

  def func1(df) {
    var df2 = df.map(func2)
    df2.cache() // 持久化中间结果
    func3(df2)
  }
  

• 移除无用的Transformation：既然df3是空的，df3.map(func2)属于无意义操作，直接删除这部分代码，避免触发不必要的计算。

内容的提问来源于stack exchange，提问作者Abhishek Agarwal