我先讲一个自己踩过的坑。早些年做数据清洗需要统计一批交易日志里到底有多少条异常记录。我按写单机代码的习惯在Driver端定义了一个普通变量count然后在map算子里写count 1跑完一看结果还是0。当时我盯着控制台愣住了——明明每一条日志都遍历到了为什么计数没变查了半天才明白问题不在业务逻辑而在Spark的分布式执行模型。这个坑就是这篇文章的主角Spark累加器Accumulator要解决的典型问题。如果你刚接触Spark或者已经写了不少RDD/DataFrame代码但一直没搞懂“为什么我在Driver定义的变量在Executor里改了没用”这篇文章就是为你准备的。我会从分布式变量隔离的原理讲起把累加器的定义、设计特点、实操代码、自定义实现、以及我踩过的各种坑一次讲透。内容不绕弯子全程按我实际排查问题的思路来。1. 先说为什么分布式环境下普通变量为什么会“失效”1.1 Driver端和Executor端之间存在一道看不见的墙Spark应用跑起来之后进程是分开的。一个Driver进程负责解析代码、生成任务、调度执行多个Executor进程负责真正干活——跑map、filter、reduce这些算子。关键点在于Driver里的普通变量在发送给Executor时会做序列化拷贝。也就是说你在Driver端写var count 0 rdd.map { x count 1 x }这行代码被序列化分发到每个Executor时count已经不是同一个变量了而是每个Task任务里的一个独立副本。每个副本在自己的Task里加加减减算倒是算了但算完就随着Task结束被抛弃没有任何机制把结果回传给Driver。最后你看看Driver端那个count还是初始值0。生活化的类比你让10个兼职人员每人拿一个小本子去统计某个路口一天的车流量要求他们把结果记在自己的本子上第二天你只看自己手里那个本子——你当然什么都看不到。累加器就是那个“大家共用一个总账本谁统计完谁把数字加起来写到总账上”的机制。1.2 为什么不能用普通变量加锁解决分布式环境的网络与性能约束有人可能会问既然Executor算完不传回来那让每个Executor把结果通过网络传回Driver不就行了理论上确实是这样但问题在于Spark的Task数量可能成千上万每个Task都回传一个小结果网络压力非常大而且如果Task中途失败你还要处理“哪些算过、哪些没算过”的问题复杂度会爆炸。更关键的约束是Task运行在分布式环境下它们之间没有共享内存。如果让所有Task并发修改同一个变量必须引入分布式锁性能代价极高完全违背Spark“移动计算而非移动数据”的设计哲学。所以Spark给出了一个专门方案累加器。它把“分布式环境下做全局计数器/汇总器”这件事封装起来让用户写起来像操作普通变量一样简单底层则处理好跨节点通信、失败重试等一系列问题。2. 累加器到底是什么分布式场景下的“只写共享变量”2.1 累加器的核心机制累加器Accumulator是Spark提供的一种分布式共享变量专用于跨Task、跨Stage地做累加操作。它的使用规则非常清晰Driver端创建你在Driver上调用SparkContext.longAccumulator或doubleAccumulator等方法创建。Executor端只写在Task里调用accumulator.add(1)或accumulator 1累加值。Driver端读值所有Task执行完成之后在Driver端调用accumulator.value获取最终结果。底层实现大致是两个阶段每个Task运行时会维护一份累加器的本地副本Task成功结束Spark会把这份副本的累加值通过网络送回Driver统一合并。这个过程对用户是透明的你只需要关心“创建、累加、读值”三个动作。// 最简单的使用方式 val total sc.longAccumulator(total) rdd.foreach { x total.add(1L) } println(total.value)2.2 三种共享变量选型对比什么时候该用累加器很多初学者分不清累加器和广播变量Broadcast Variable的区别这里直接给一张对比表维度累加器广播变量普通Driver变量数据流向Executor → Driver单向汇总Driver → Executor只读分发Driver → Executor拷贝分发是否可修改Executor只能累加不读Executor只读不可修改Executor可改但不回传典型场景计数、求和、指标统计大字典、配置表、模型参数小型只读参数少量是否有容错机制有Task级精确有缓存机制无结论很简单你需要“全局汇总”的时候用累加器你需要“共享只读大对象”的时候用广播变量。两者互补并不冲突。我见过不少项目里同时用广播变量传配置字典、用累加器统计脏数据条数各司其职。3. 累加器的三大特点每一个都值得你记在笔记里3.1 特点一只写不读Executor端永远看不到最终值累加器最反直觉的一点是Executor端可以“写”累加但不可以“读”。如果你在Executor端调用accumulator.value拿到的只是这个Task本地副本的初始值或部分累加值而不是全局最终结果。我见过一些人踩的坑在map算子内部判断“如果accumulator.value大于某个阈值就执行特殊逻辑”。这个写法结果完全不可控——因为每个Task看到的只是一个局部副本甚至可能是0。正确的做法是所有读操作都放在Driver端等Action触发、任务全部跑完后统一读。这个设计不是拍脑袋定的。如果允许Executor端实时读取全局值意味着每个Task每次读都要向Driver发起远程请求不仅网络开销巨大还会引入读写一致性问题。Spark选择把“全局值”的读取收敛到Driver端换来的是简单和可靠。3.2 特点二全局唯一同一个累加器贯穿整个作业你在Driver端创建了一个累加器那么整个Spark作业里所有Stage、所有Task对这个累加器的累加操作都是汇总到一个逻辑实体上。比如一个作业先做过滤、再做分组、再做输出三个阶段里你都可以往同一个累加器上累加数据。这个“全局唯一”特性在实际工作中非常有用。举个例子我在做日志清洗时经常同时统计三件事输入总条数、异常记录条数、过滤后有效条数。不需要写一堆map之后还得各自count直接创建三个累加器在同一个转换逻辑里顺手加一下就行。val totalCount sc.longAccumulator(total) val errorCount sc.longAccumulator(error) val validCount sc.longAccumulator(valid) logs.foreach { record totalCount.add(1L) if (record.contains(ERROR)) { errorCount.add(1L) } else { validCount.add(1L) } }一个Action跑完三个值同时拿到省掉多次扫描数据的时间。3.3 特点三容错机制Task重试不会重复累加Stage重算会重复累加累加器的容错机制是最容易出问题、也最值得讲清楚的地方。Spark为累加器设计了一套TASK级别的“精确一次”机制。具体来说每个Task在执行过程中维护一份独立的累加器副本只有Task成功完成时这份副本值才会被回传Driver合并。如果Task执行失败被重试之前那个失败Task的累加副本会被直接丢弃不会重复累加到全局结果里。这一点在很多资料里被称为“Task级别不重复计数”。但是这里有一个重要例外如果Stage因为数据丢失被完整重算那么整个Stage里的累加器会被重新执行一遍。比如某个RDD分区数据在缓存中被清除了后续Action需要重新计算这时之前已经执行过的累加操作会再执行一次累加结果就会偏大。只看官方文档容易忽略这个细节。在实际做数据统计时如果你依赖累加器做精确的金额汇总遇到Stage重算就会得到偏大的数。所以我在项目里养成了一个习惯累加器主要用于计数、条数统计、调试信息收集这类“允许一定冗余”的场景涉及金额等精确数值的汇总尽量用reduce或DataFrame的聚合算子来做。这样各取所长避开累加器重算的软肋。4. 实操演示用累加器统计JSON日志中的异常次数4.1 完整代码与运行过程下面这段代码基于Spark读取JSON日志正好对应很多人在搜的“spark中读取json”场景。假设我们有一批交易日志每条日志包含级别、消息、时间等字段我要统计其中levelERROR的记录数量。import org.apache.spark.sql.SparkSession object AccumulatorDemo { def main(args: Array[String]): Unit { val spark SparkSession.builder() .appName(AccumulatorDemo) .master(local[*]) .getOrCreate() val sc spark.sparkContext // 1. 在Driver端创建累加器 val errorAccum sc.longAccumulator(errorCount) val totalAccum sc.longAccumulator(totalCount) // 2. 读取JSON格式日志文件 val logs spark.read.json(hdfs:///data/logs/2024-09-01/*.json) val logRdd logs.rdd // 3. 在转换算子中累加 logRdd.foreach { row totalAccum.add(1L) val level row.getAs[String](level) if (ERROR.equalsIgnoreCase(level)) { errorAccum.add(1L) } } // 4. 在Driver端读取最终值 println(s总日志条数: ${totalAccum.value}) println(sERROR日志条数: ${errorAccum.value}) spark.stop() } }这段代码的意思是totalAccum记录处理的总行数errorAccum记录级别为ERROR的行数。foreach是一个Action操作它触发真正的执行Task在各Executor上跑完自动把结果回传给Driver。最终打印出来的就是准确的总数和异常数。4.2 对照实验普通变量到底丢了多少为了加深理解我还专门写过一段对照代码用普通变量替换累加器var errorCount 0 logRdd.foreach { row val level row.getAs[String](level) if (ERROR.equalsIgnoreCase(level)) { errorCount 1 // 这里改的是Task本地副本 } } println(sERROR条数: $errorCount) // 永远是0运行结果永远是0和你是不是本地模式无关。即便在local[*]模式下跑Driver和Executor也是分离的执行环境变量修改依旧不会回传。你还可以换一种写法把它放进RDD的map里而不是foreach里logRdd.map { row if (isError(row)) errorCount 1 row }.collect()这同样拿不到正确结果。因为Transform操作是惰性的RDD的map很可能不会真正执行没有Action触发或者因为collect触发执行了但修改同样留在Executor副本里。记住分布式环境里通过闭包捕获到Executor端的变量改了多少都是“自说自话”只有累加器这种专门的设计才能把结果汇总回来。5. 自定义累加器AccumulatorV2高级用法5.1 内置累加器的局限Spark内置了三种比较基础的累加器分别对应Long、Double、Collection。它们的局限很明显longAccumulator只能做整数的加和。doubleAccumulator只能做浮点数的加和。collectionAccumulator可以收集元素到一个集合但集合内不能做自定义合并规则。如果我只是统计“数量”和“和值”内置累加器够用。但业务往往没这么简单。举个例子我参与过一个农产品价格数据分析的项目需要按品类统计“价格异常次数”即每种农产品分别有多少条记录超出了正常波动区间。如果用内置累加器只能统计总数无法按品类拆分。这时候就需要自定义累加器。5.2 手写一个按品类统计的自定义累加器Spark提供了抽象类AccumulatorV2[IN, OUT]第一个泛型是每个Task往里累加的数据类型第二个泛型是Driver端最终读到的结果类型。下面是我写的一个按品类累计异常次数的累加器import org.apache.spark.util.AccumulatorV2 import scala.collection.mutable class CategoryAccumulator extends AccumulatorV2[String, Map[String, Long]] { private val counts mutable.Map[String, Long]() // 判断当前累加器是否为空 override def isZero: Boolean counts.isEmpty // 拷贝一份给Task用 override def copy(): CategoryAccumulator { val newAcc new CategoryAccumulator counts.foreach { case (k, v) newAcc.counts.put(k, v) } newAcc } // 重置为初始状态 override def reset(): Unit counts.clear() // Task端每处理一条记录调用一次 override def add(category: String): Unit { counts.put(category, counts.getOrElse(category, 0L) 1L) } // 多个Task的副本合并关键中的关键 override def merge(other: CategoryAccumulator): Unit { other.counts.foreach { case (k, v) counts.put(k, counts.getOrElse(k, 0L) v) } } // Driver端读取的最终结果 override def value: Map[String, Long] counts.toMap }使用方法和内置累加器几乎一模一样val acc sc.register(new CategoryAccumulator, categoryAcc) rdd.foreach { row val category row.getAs[String](category) val price row.getAs[Double](price) val avgPrice getAvgPrice(category) if (price avgPrice * 1.2) { acc.add(category) // 这里传入的是品类名 } } acc.value.foreach { case (k, v) println(s$k: $v) }value返回一个Map[String, Long]每一项对应一个品类的异常次数。相比内置累加器这个方式一次就能汇总出多维度的统计结果不用按品类写一堆filter再分别count。5.3 自定义累加器的关键细节写自定义累加器最容易犯错的地方是merge方法。我举个例子如果漏掉merge或者写成“直接覆盖”那么多个Task的结果就只剩下最后一个Task的累加值最终结果会严重偏小。之所以会这样是因为Spark在Driver端拿到每个Task回传的副本后必须通过reduce方式把它们合并成一个最终值merge就是那个合并逻辑。你还需要注意以下几点两个泛型参数别搞反AccumulatorV2[IN, OUT]IN是Task端add方法的输入类型OUT是Driver端value方法的输出类型。copy和reset一般配对出现copy方法用于为每个Task复制一份初始状态reset用于重置。如果这两个方法实现不对Task执行时可能带着之前的状态累加导致结果翻倍。序列化问题自定义累加器最终要在Driver和Executor之间传输所以不要塞入不可序列化的对象比如数据库连接、文件句柄否则会报序列化异常。注册之后才能用在Driver端创建自定义累加器时必须调用sc.register(acc, 名字)否则在集群模式下会报“未注册”相关的错误。6. 累加器使用中的常见问题与排查记录6.1 executor端读到的累加器值永远是0现象在map或foreach内部打印累加器的value发现一直输出0。原因正如前面所说Executor端拿到的value只是Task本地副本的值。对大多数Task来说刚开始本地副本还是0等你累加完之后再去读时才有一点局部值但绝对不是全局值。解决思路不要在Executor端读值。如果需要根据累加结果做进一步判断请把最终结果保存到Driver端变量里再做后续逻辑。6.2 同一个累加器在transform算子中被重复累加现象程序跑完累加器结果比预期大了好几倍而且每次运行结果不一致。原因map、filter这类Transform是惰性的同一个RDD如果被多个Action共用或者因为persist缓存失效导致重新计算里面的累加代码会执行多次。一个典型的错误写法是val rdd2 rdd1.map { x acc.add(1L) x 1 } rdd2.count() // 第一次触发累加一次 rdd2.collect() // 如果rdd1未缓存再一次从头计算累加两次如果rdd1没有persist(StorageLevel.MEMORY_ONLY)第二次Action会重新执行整个血缘链导致累加器加了两遍。排查方法检查累加器写入的算子到底是Transform还是Action。如果必须写在Transform里确保上游RDD被persist或者cache住了避免重复计算。6.3 stage重算导致累加器结果偏大现象集群某个节点故障后作业重跑或者动态资源调整导致分区数变化最终累加值和预期不符。原因我们前面讲过Spark的容错机制。Task失败重试不会重复累加Task级局部值但Stage级重算会让所有Task重来一遍自然会把累加器再加一遍。如果作业发生了Executor丢失、节点故障等异常累加器的精确性就无法保证。排查方法查看Spark UI上是不是有Stage Retry重试的记录尤其是因为FetchFailed引起的Shuffle重试。这类场景下累加器只能作为“粗粒度指标参考”不宜用做精确的财务统计。我实测下来Spark UI的“Accumulator”标签页会显示每个Stage的累加值对照着就能看出哪个Stage重算了。6.4 自定义累加器合并结果错误现象自定义累加器最后拿到的value比预期少或者根本是0。原因多半是merge方法写了覆盖逻辑。比如override def merge(other: MyAcc): Unit { this.count other.count // 错误示范覆盖而不是相加 }这会导致Driver端合并多个Task副本时后合并的覆盖先合并的最后只剩一个Task的局部值。排查方法先在本地小数据上用两个分区跑人为制造多个Task打印每个Task回传的局部快照再看合并后的最终值。本地模式的local[2]就能复现这个情况。确认merge是“两两相加”而不是“赋值覆盖”问题就解决了。写在最后的经验之谈我自己用累加器的习惯是比较“克制”的。和项目相关的监控指标比如输入数据量、异常率、解析失败条数、按品类统计的分布我用累加器做得很顺手但涉及金额、库存这类需要精确到分的数值我一律用DataFrame的内置聚合算子来算宁可多写几行代码也不想在Stage重算时被坑。还有个细节是在YARN集群模式下自定义累加器如果带中文名有些旧版集群在Spark UI上会显示乱码建议名称只用英文和下划线。另外凡是需要用累加器结果触发后续逻辑的场景一定要等Action完成后在Driver端取value再判断不要试图在Task里做判断。写这篇东西的初衷其实就是想让你不用再经历我当年的困惑。分布式系统里“共享一个变量”这件事远没有看起来那么简单累加器帮你封装了跨节点通信、合并、容错这些脏活但它的边界只写不读、Stage重算不精确也需要你心里有数。当你真正理解了这些边界再写Spark算子时就会更笃定——知道什么选择是合理的也知道踩坑之后该怎么排查。