为什么RDD执行foreach操作后Driver端的ArrayBuffer为空？如何解决？

• Spark采用分节点分布式执行模型，算子（如foreach）内的逻辑会作为闭包分发到各个Executor进程执行。闭包引用的Driver端变量会被序列化后拷贝到Executor本地，算子内所有修改操作都是针对Executor本地的副本进行，完全不会同步回Driver端的原始变量。
• 你在foreach内部看到的长度递增打印，实际上是各个Executor上本地副本的变化，Driver端初始定义的calcUsers从未被修改，因此最终打印始终为空。

根据你的使用场景选择对应方案：

• 如果你需要将所有用户ID收集到Driver端处理：直接使用collect算子拉取结果即可

import scala.collection.mutable.ArrayBuffer

val calcUsers: ArrayBuffer[Int] = nRDD
  .map(item => item.split(" ")(1).toInt)
  .collect()
  .to(ArrayBuffer)
println("calcUsers", calcUsers, calcUsers.length)

• 如果你只需要统计全局用户数等聚合指标：优先使用Spark自带的累加器Accumulator，这是Spark官方提供的唯一支持跨节点全局累加的变量类型

import org.apache.spark.util.LongAccumulator

// 初始化累加器，第二个参数为累加器名称，可在Spark UI查看
val userCountAcc: LongAccumulator = sc.longAccumulator("user_count")
nRDD.foreach(item => {
  val currUserId = item.split(" ")(1).toInt
  userCountAcc.add(1)
  // 其他业务逻辑
})
// 只有Driver端可以读取累加器的最终值
println("用户总数", userCountAcc.value)

注意：禁止在Spark算子内部修改Driver端定义的普通变量，Spark没有提供普通变量的跨节点同步机制，此类用法不符合Spark的执行模型，一定会出现预期外的结果。

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