先说一个几乎所有Spark初学者都会在第一个周末遇到的翻车现场。我们想知道一次批处理到底处理了多少行数据于是很自然地在driver端写了一个var count 0打算在一个map或foreach里让count加一等作业跑完直接打印count。结果完全出乎意料任务正常结束count还是0。更诡异的是你翻executor日志每个task里的count确实在涨但每个task都是从0开始涨的涨到一定数量就停了driver这边纹丝不动。这个现象背后的原因就是Spark累加器存在的意义。我当时第一次遇到也是一脸懵后来老老实实把累加器用了起来包括LongAccumulator、DoubleAccumulator再到后来自己继承AccumulatorV2写自定义累加器才慢慢把这块彻底吃透。这篇就把累加器是什么、为什么需要、有哪些特点一次说清楚顺便带上我在生产环境里踩过的坑和自定义累加器的完整案例。1. 为什么分布式场景下不能靠共享变量计数1.1 复现一个经典的翻车现场先看一段最典型的错误代码应该能引起不少人的共鸣var count 0 val lines sc.textFile(hdfs:///logs/app.log) lines.foreach { line count 1 } println(s总行数: $count)你以为这是一段非常自然的代码driver端定义一个计数器executor处理每条数据时加一最后driver打印总数。但实际跑完count永远等于0。你把println改成下面这样再跑一次lines.foreach { line count 1 println(stask内count$count) }这时候你会看到executor的stdout里count确实有在增长但每个task的count都是从0开始、到该task处理的行数为止从来不会是全集群的累计值。也就是说每个executor/task都在维护一个属于自己的本地副本各改各的改完也不会有人把这些副本汇总回driver。1.2 闭包序列化task副本与变量回传的缺失为什么会出现这种情况核心在于Spark的分布式执行模型。当你在driver端写出lines.foreach { line count 1 }时Spark需要把这个匿名函数闭包发送到各个executor上执行。发送之前Spark会序列化闭包中引用到的所有外部变量也就是那个count。此时count的值是多少0。这个0被拷贝进了闭包闭包被序列化成字节码通过网络分发到每个executor再反序列化生成一个个task实例。于是每个task拿到的count都是driver端那个count在发送时刻的一个快照副本。task里的count 1修改的完全是JVM进程内、属于自己的一份局部变量它跟driver进程里的那个count没有任何引用关系。任务跑完后task的结果Result里会携带处理后的数据、累加器更新等元信息但绝对不会携带一个普通局部变量的最后值回传给driver。这就是count归零的根本原因。用一句话概括在分布式计算里跨进程修改一个普通变量没有任何意义因为进程之间没有共享内存也没有隐式的变量回传通道。1.3 没有累加器时你可能被迫使用的三种笨办法没有累加器这个官方通道大家通常会硬想出几种替代方案但都各有各的难受先把数据collect回driver再数。比如lines.collect().length。数据量小的时候一点问题没有但批处理动辄几千万上亿条collect等于把所有数据一次性拉到driver端轻则OOM重则直接把driver搞崩溃。你要是铁了心用collect做海量统计内存和GC一定先扛不住。用reduce或aggregate做精确聚合。为了统计一个总数你得先把RDD映射成(key, 1)的结构再reduceByKey(_ _)最后collect回来。这个方案结果精确、逻辑也正确但为了一个计数器要写这么多样板代码还要引入一次shuffle在只想做作业内监控指标的场景下显得非常笨重。写外部存储做中间计数。比如在每个task里把count写到Redis或数据库driver最后去读。这个方案引入了外部依赖多了一次写库开销而且在task失败重试时你还得自己处理幂等否则计数翻倍。纯粹为了数行数去搭一套Redis属于杀鸡用牛刀。你会发现这些替代方案要么性能差、要么代码繁琐、要么依赖外部组件。而Spark之所以专门提供累加器就是因为它想用一种轻量的方式解决分布式环境下的聚合统计和作业内监控这个通用需求。2. 累加器的运行机制从一次add到merge的完整路径2.1 累加器到底长什么样AccumulatorV2核心接口Spark 2.0之后累加器的底层统一抽象为AccumulatorV2[IN, OUT]。这是一个抽象类你写的自定义累加器基本就是继承它。先看核心方法abstract class AccumulatorV2[IN, OUT] extends Serializable { // 将累加器重置为零值driver端和task副本都会调用 def reset(): Unit // 向累加器写入一个值IN是输入类型 def add(v: IN): Unit // 把另一个累加器的值合并到当前累加器 def merge(other: AccumulatorV2[IN, OUT]): Unit // 返回当前累加器的最终值OUT是输出类型 def value: OUT // 复制当前累加器的副本每个task执行前会调用 def copy(): AccumulatorV2[IN, OUT] // 判断累加器是否处于零状态 def isZero: Boolean }内建的LongAccumulator、DoubleAccumulator就是对这几个方法的最简单实现add做加法merge做累加求和value返回最后的总数。理解这几个方法就理解了累加器全部的数据流。2.2 一次完整累加从driver注册到task结果回传把一次累加计数的完整生命周期拆开看一共六步driver端创建并注册。你调用sc.longAccumulator(counter)或自己new一个累加器再sc.register(acc, name)时Spark会把累加器的初始状态注册到SparkContext里。注册的意义在于Spark需要追踪这个累加器便于后续在task结果中携带它的更新、在UI上展示、以及在某些重试场景下恢复初始状态。闭包序列化分发。当RDD算子引用了这个累加器对象Spark会像序列化普通变量一样把它连同闭包一起发给executor。每个executor拿到的又是一个副本。task执行前copy一份。更准确地说每个task在执行前会调用copy()方法把累加器复制一份独立的实例。task只在这个副本上执行add多个task互不干扰这也是累加器在并发环境下能保持线程安全的原因之一。task本地执行add。你在map或foreach里调用acc.add(1)实际上只更新了task自己的那个副本。task内可以调用任意多次add完全本地操作没有任何网络通信开销这就是累加器性能高的核心原因。task完成结果回传driver。task执行完毕后Spark会把累加器的本地副本值打包进task的TaskResult中随结果一起回传给driver。注意这里回传的不是累加过程中产生的每一个值而是经过本地合并后的一个快照。driver端执行merge。driver收到每一个task的结果后取出里面的累加器副本调用全局累加器的merge(other)方法把task的结果合并进全局状态。所有task的merge都完成之后你就能通过acc.value拿到统计结果了。整个流程用一句话概括task端只负责adddriver端负责merge中间靠task结果回传作为桥梁。2.3 内建累加器与老版本API的差异平时最常用的两个内建累加器是sc.longAccumulator和sc.doubleAccumulator分别对应LongAccumulator和DoubleAccumulator只支持数值累加。它们的好处是简单、快、不需要任何自定义代码。但很多老教程里还在用这样的写法val counter sc.accumulator(0, counter)这是Spark 1.x时代的API底层基于AccumulatorParam。如果你今天还在用Spark 2.0以上的版本建议直接换成sc.longAccumulator别再看老代码了。sc.accumulator在2.0之后已经标记为废弃虽然还能编译通过但类型体系、性能设计和UI展示都不如新版接口完善。另外提一嘴累加器和广播变量经常被放在一起对比广播变量是只读共享把大变量广播到executor供所有task读取累加器是只写共享task端只能写driver端只能读最终结果。两者一个负责下发放置数据、一个负责收集计算结果配合使用时可以覆盖很多分布式编程场景。3. 累加器的四个关键特点惰性触发、只写不读与重复计算很多文章介绍累加器特点时就写一句累加器是Spark提供的累加变量这远远不够。实操中真正决定你用不用、怎么用累加器的是下面这四个特点。前两个是表面特性后两个才是真正的深坑。3.1 特点一行动算子触发生效累加器是懒惰的这是Spark执行模型的一部分RDD上的transformation算子都是懒执行的累加器的更新也遵循这个规则。看这段代码val acc sc.longAccumulator(counter) val mapped rdd.map { x acc.add(1) x } // 只执行到这里不触发action println(acc.value) // 输出0map是transformation它只是构建了血统关系并没有真正跑起来。累加器的add一次都不会执行driver的acc.value自然还是0。只有当你调用count、collect、saveAsTextFile、foreach这类行动算子时job才真正开始执行add才会被调用。这个特性带来的一个陷阱是如果同一个RDD被多次行动算子触发且每次执行都会重新计算之前的transformation那累加器就会累加多次。比如val rdd someSource.map { x acc.add(1); x } rdd.count() // acc第一次累加 rdd.count() // acc再次累加值翻倍即使你用了rdd.cache()如果executor内存不够导致分区被提前清理或者cache之后源数据变了Spark仍然可能重算。因此在判断累加器数值对不对之前先确认这个RDD到底被行动算子执行了多少次。3.2 特点二executor端只写不读task里读到的永远不是全局值累加器在task里只能执行add你如果尝试在executor端读取acc.value读到的只是task本地副本的值通常是初始值或这个task自己刚刚累加的结果绝不是driver端累积了所有task之后的全量值。rdd.foreach { x acc.add(1) // 这里读到的value几乎可以确定是错误的、不完整的 if (acc.value 1000) { // 你永远等不到这个分支触发 } }所以在map或foreach里写当累加器达到某个阈值就做某件事这种逻辑是完全行不通的。原因也好理解如果每个task都要读取driver端的全局实时值那每次add都得做一次网络往返累加器就失去性能优势了而且多task并发执行时全局值时刻在变你读到某个值根本不能代表稳定的全局状态做控制流决策反而会得到错误结论。3.3 特点三task重试和stage重算会造成重复累加这是最大的坑这个是累加器在生产环境中最容易翻车的点没有之一。Spark的容错机制是通过重算来恢复。一个task因为节点宕机、executor失联、内存溢出等原因失败了TaskScheduler会把它重新调度到其他executor上执行。如果一个stage的shuffle输出文件丢失Spark会把这个stage及下游stage重新计算。在这些重试和重算的过程中累加器的add会被再次执行但是Spark不会自动把之前已经累加进去的数值回滚。举个例子一个task本来已经执行成功add了500次结果回传给driver了。但driver后来发现另一个stage需要重算这个task又被重新执行了一次又add了500次。最终driver上累加器的值就变成了1000而不是真实的500。这也是为什么Spark官方文档明确提醒对于需要在失败重试时保证结果精确的作业累加器不做这种保证。累加器在容错语义上是best-effort的它能满足监控、统计、诊断这类多算一点点可以接受的场景但不适合充当业务结果的计算工具。我遇到过一个实际案例某个数仓任务用累加器统计过滤掉的异常行数某天凌晨集群节点出问题一批task重试了3轮当天上报的异常行数直接翻了快3倍看起来像数据突然恶化了实际上只是重试叠加而已。3.4 特点四merge逻辑可自定义但设计时必须考虑交换性和结合性累加器merge(other)是用来合并两个累加器状态的。对于数值累加器merge就是加法天然满足交换律和结合律。但自定义累加器时你完全可以定义自己的合并逻辑比如并集、最大值、最值集合等。这里有一个隐含要求多个task的merge顺序是不确定的。driver可能先收到task A的结果再收到task B的结果也可能反过来执行merge时是逐个合并进全局累加器全局状态不保证按某个固定顺序合并。因此你的merge逻辑必须是可交换的、可结合的。比如取最大值求和取集合并集都满足但求平均值就不满足——A和B先合并再与C合并和B与C先合并再与A合并结果一样吗不一定。我见过有人想用累加器做平均值统计写出来的merge是(this.sum other.sum) / (this.count other.count)这就是典型的错误设计最终结果完全取决于merge顺序。遇到这类需求正确做法是累加器里只维护sum和count两个值value方法里最后再算平均值merge时只合并原始sum和count。3.5 特点小结一张表看清累加器的边界特点具体表现实操注意点惰性执行只有行动算子触发job时才更新同一RDD被执行多次累加器会重复累加只写不读task端只能add读.value得到的是本地副本不能在task里基于累加器值做控制流容错重算task重试、stage重算会重复执行add不适合做业务精确计量适合监控诊断自定义merge多个task结果的合并顺序不确定merge要满足交换律和结合律4. 自定义累加器实战用数据质量统计把原理跑通4.1 为什么需要自定义内置累加器的能力边界LongAccumulator和DoubleAccumulator只能做简单的数值累加。但真实场景里我们经常需要在一次作业运行过程中同时统计多种指标日志数据里某个字段的空值数量数值字段为负数的记录数非法枚举值的数量某几个字段的最大值、最小值过滤条件丢弃的各类原因计数如果每次都要单独定义一个LongAccumulator那会累积出一堆变量register代码和驱动端读取代码都变得很啰嗦。更合理的做法是把这些统计项封装进一个自定义累加器让add接收一条记录value返回一个完整的统计报告。4.2 完整代码一个数据质量累加器下面是我在数据清洗作业里用过的思路简化后分享出来。它实现的功能是统计输入日志中空值数、负数值、非法值数以及正常记录数。import org.apache.spark.util.AccumulatorV2 import scala.collection.mutable class DataQualityAccumulator extends AccumulatorV2[String, Map[String, Long]] { private val counters mutable.Map[String, Long]() override def isZero: Boolean counters.isEmpty override def copy(): DataQualityAccumulator { val acc new DataQualityAccumulator counters.foreach { case (k, v) acc.counters.put(k, v) } acc } override def reset(): Unit counters.clear() override def add(record: String): Unit { // 这里按实际解析结果决定统计哪个维度 record match { case uid_null counters(uid_null) counters.getOrElse(uid_null, 0L) 1L case negative_time counters(negative_time) counters.getOrElse(negative_time, 0L) 1L case invalid_city counters(invalid_city) counters.getOrElse(invalid_city, 0L) 1L case valid counters(valid) counters.getOrElse(valid, 0L) 1L case _ counters(other) counters.getOrElse(other, 0L) 1L } } override def merge(other: AccumulatorV2[String, Map[String, Long]]): Unit { other.value.foreach { case (k, v) counters(k) counters.getOrElse(k, 0L) v } } override def value: Map[String, Long] counters.toMap }使用方式也很简单val qualityAcc new DataQualityAccumulator sc.register(qualityAcc, data-quality) val df spark.read.json(/data/logs/2024-06-01) df.foreachPartition { iter iter.foreach { row val uid row.getAs[String](uid) val time row.getAs[String](event_time) val city row.getAs[String](city) if (uid null) qualityAcc.add(uid_null) else if (time null || time.toLong 0) qualityAcc.add(negative_time) else if (city null || city.isEmpty) qualityAcc.add(invalid_city) else qualityAcc.add(valid) } } val report qualityAcc.value println(s数据质量报告: $report)这里我用了foreachPartition而不是foreach目的很明确在每个分区内复用迭代逻辑减少add调用的外层开销。不过要注意无论foreach还是foreachPartition本质都是行动算子累加器都会正常生效。4.3 代码之外的四个注意点第一累加器类必须可序列化。它会被序列化后分发给executor所以内部不要持有不可序列化的对象比如没实现的连接池、非序列化的日志框架等。一个简单的检查方法是在类声明里继承SerializableAccumulatorV2已经继承了并且所有字段都用mutable.Map、Long这类可序列化类型。第二value方法要返回不可变快照。我上面返回的是counters.toMap而不是直接返回counters这个mutable.Map。如果直接返回可变Mapdriver端读完再修改很容易污染全局累加器状态引发并发问题。第三copy方法要真正复制状态。每个task执行前都会调用copy如果copy返回的是this本身或者共享同一个Maptask之间的add会互相污染最终统计结果会乱成一锅粥。copy里一定是new一个新对象把当前counters的键值逐个拷过去。第四严禁往累加器里塞明细数据。有些新手会想用自定义累加器收集所有非法行把原始数据都塞进累加器的集合里。这个想法很危险每个task都会把非法的原始数据回传driverdriver内存会被瞬时打爆。正确姿势是只统计计数原始明细写在executor本地或通过旁路日志输出绝不能让driver当数据汇集中心。4.4 为什么一定要sc.register可能有同学会问我不调sc.register累加器也能正常add、value啊register有什么用register至少有三层意义让Spark把累加器注册到SparkContext的累加器列表中driver可以随时拿到它的引用UI界面的Accumulators标签页里也会显示它的当前值方便你实时观察进度。注册是Spark恢复机制的一部分。在部分任务重试场景下Spark需要知道哪些累加器需要重置或重建注册过的累加器才能被正确管理。不注册的累加器在Web UI里没有名字、没有历史记录出了事你连值班日志都没得查。所以自定义累加器写完之后第一件事就是sc.register(acc, 可读名字)这个名字会显示在Spark UI里排查问题全靠它。5. 生产环境里的选型判断累加器、reduceByKey和Metrics谁更合适5.1 先看一张对比表方案精确性性能开销适用场景累加器不保证容错精确重试会重复累加很低task内纯本地操作作业内监控、数据质量统计、调试信息reduceByKey/aggregate精确失败重算后结果正确需要shuffle有网络IO业务最终聚合结果collect回driver计算精确但数据量大有OOM风险很慢driver成为瓶颈小数据量验证、调试Spark内建Metrics精确Spark自身统计无额外代码开销shuffle字节数、输入记录数等系统监控这个表的本质区别是什么累加器是尽力而为的统计工具reduceByKey才是保证正确的计算算子。两者的定位完全不同不要把累加器当成精确计数的替代品。5.2 我在生产环境里的使用习惯经验一累加器用来做过程监控不参与业务结果计算。比如在ETL作业里我会用累加器统计每条数据被过滤的原因等作业跑完这些统计写到日志或监控面板里用于判断数据质量波动。但如果这个统计结果要进入下游报表、参与金额计算我一定会用reduceByKey或agg函数重新计算绝不直接用累加器输出。经验二把累加器统计的记录总数和rdd.count()做交叉校验。这是个非常实用的技巧。比如累加器统计出valid invalid总共N条如果rdd.count()和N对不上说明要么累加器没有覆盖所有分支要么有stage重算导致重复累加。这种校验能帮你在一开始就发现代码逻辑或容错的隐患而不是等上线之后才暴露。经验三在DataFrame上使用累加器优先配合foreachPartition。DataFrame的foreach一行一行执行效率偏低foreachPartition一次处理一个分区可以同时结合add逻辑和本地缓存减少每行的调度开销。如果你的统计逻辑比较复杂比如要解析JSON、做正则匹配一个分区里再包一层迭代器循环能省不少事。经验四不要在生产环境里用累加器往外部系统实时推数据。累加器的执行时机是task粒度的task完成才会把值传给driver不是每add一次就通知一次。如果你想实时感知处理进度应该用Streaming或Structured Streaming里的状态更新机制而不是指望累加器。5.3 最后的排查技巧让Accumulators标签页帮你写作业写到最后分享一个我一直保留的习惯。每次跑Spark作业我都会在Web UI的Accumulators标签页里盯着自定义累加器的变化。这个页面会展示每个累加器的当前值和历史值比你在代码里println要直观得多。有一次线上作业的数据量级突然异常我第一时间打开Accumulators标签页发现某个累加器的值比上一批作业翻了正好一倍而其他指标都正常。顺着这个线索查下去发现是上游新加了一个重试机制导致同一份数据被计算了两次。如果没有这个UI页面单靠代码日志排查至少得多花一小时。还有一个经验如果哪天你的累加器数值看起来莫名其妙地大先怀疑stage重算或RDD被多次执行而不是怀疑自己的代码逻辑。这个排查顺序能帮你省下很多时间尤其是集群资源不稳定、任务频繁重试的时候。累加器本身不是什么高深机制但它是理解Spark分布式执行模型的一把钥匙。把task本地add、driver统一merge、重试会重复累加这几个核心点刻在脑子里用起来就不会踩大坑。至于什么时候该自己继承AccumulatorV2我的建议很简单当你需要统计的维度超过五六个、而且想在一次作业里集中汇总时就值得动手写了。 
长河君 · 2026/10/9 3:08:29 · 阅读 1.2k Spark累加器原理与实战：从翻车现场到生产避坑指南
先说一个几乎所有Spark初学者都会在第一个周末遇到的翻车现场。我们想知道一次批处理到底处理了多少行数据于是很自然地在driver端写了一个var count 0打算在一个map或foreach里让count加一等作业跑完直接打印count。结果完全出乎意料任务正常结束count还是0。更诡异的是你翻executor日志每个task里的count确实在涨但每个task都是从0开始涨的涨到一定数量就停了driver这边纹丝不动。这个现象背后的原因就是Spark累加器存在的意义。我当时第一次遇到也是一脸懵后来老老实实把累加器用了起来包括LongAccumulator、DoubleAccumulator再到后来自己继承AccumulatorV2写自定义累加器才慢慢把这块彻底吃透。这篇就把累加器是什么、为什么需要、有哪些特点一次说清楚顺便带上我在生产环境里踩过的坑和自定义累加器的完整案例。1. 为什么分布式场景下不能靠共享变量计数1.1 复现一个经典的翻车现场先看一段最典型的错误代码应该能引起不少人的共鸣var count 0 val lines sc.textFile(hdfs:///logs/app.log) lines.foreach { line count 1 } println(s总行数: $count)你以为这是一段非常自然的代码driver端定义一个计数器executor处理每条数据时加一最后driver打印总数。但实际跑完count永远等于0。你把println改成下面这样再跑一次lines.foreach { line count 1 println(stask内count$count) }这时候你会看到executor的stdout里count确实有在增长但每个task的count都是从0开始、到该task处理的行数为止从来不会是全集群的累计值。也就是说每个executor/task都在维护一个属于自己的本地副本各改各的改完也不会有人把这些副本汇总回driver。1.2 闭包序列化task副本与变量回传的缺失为什么会出现这种情况核心在于Spark的分布式执行模型。当你在driver端写出lines.foreach { line count 1 }时Spark需要把这个匿名函数闭包发送到各个executor上执行。发送之前Spark会序列化闭包中引用到的所有外部变量也就是那个count。此时count的值是多少0。这个0被拷贝进了闭包闭包被序列化成字节码通过网络分发到每个executor再反序列化生成一个个task实例。于是每个task拿到的count都是driver端那个count在发送时刻的一个快照副本。task里的count 1修改的完全是JVM进程内、属于自己的一份局部变量它跟driver进程里的那个count没有任何引用关系。任务跑完后task的结果Result里会携带处理后的数据、累加器更新等元信息但绝对不会携带一个普通局部变量的最后值回传给driver。这就是count归零的根本原因。用一句话概括在分布式计算里跨进程修改一个普通变量没有任何意义因为进程之间没有共享内存也没有隐式的变量回传通道。1.3 没有累加器时你可能被迫使用的三种笨办法没有累加器这个官方通道大家通常会硬想出几种替代方案但都各有各的难受先把数据collect回driver再数。比如lines.collect().length。数据量小的时候一点问题没有但批处理动辄几千万上亿条collect等于把所有数据一次性拉到driver端轻则OOM重则直接把driver搞崩溃。你要是铁了心用collect做海量统计内存和GC一定先扛不住。用reduce或aggregate做精确聚合。为了统计一个总数你得先把RDD映射成(key, 1)的结构再reduceByKey(_ _)最后collect回来。这个方案结果精确、逻辑也正确但为了一个计数器要写这么多样板代码还要引入一次shuffle在只想做作业内监控指标的场景下显得非常笨重。写外部存储做中间计数。比如在每个task里把count写到Redis或数据库driver最后去读。这个方案引入了外部依赖多了一次写库开销而且在task失败重试时你还得自己处理幂等否则计数翻倍。纯粹为了数行数去搭一套Redis属于杀鸡用牛刀。你会发现这些替代方案要么性能差、要么代码繁琐、要么依赖外部组件。而Spark之所以专门提供累加器就是因为它想用一种轻量的方式解决分布式环境下的聚合统计和作业内监控这个通用需求。2. 累加器的运行机制从一次add到merge的完整路径2.1 累加器到底长什么样AccumulatorV2核心接口Spark 2.0之后累加器的底层统一抽象为AccumulatorV2[IN, OUT]。这是一个抽象类你写的自定义累加器基本就是继承它。先看核心方法abstract class AccumulatorV2[IN, OUT] extends Serializable { // 将累加器重置为零值driver端和task副本都会调用 def reset(): Unit // 向累加器写入一个值IN是输入类型 def add(v: IN): Unit // 把另一个累加器的值合并到当前累加器 def merge(other: AccumulatorV2[IN, OUT]): Unit // 返回当前累加器的最终值OUT是输出类型 def value: OUT // 复制当前累加器的副本每个task执行前会调用 def copy(): AccumulatorV2[IN, OUT] // 判断累加器是否处于零状态 def isZero: Boolean }内建的LongAccumulator、DoubleAccumulator就是对这几个方法的最简单实现add做加法merge做累加求和value返回最后的总数。理解这几个方法就理解了累加器全部的数据流。2.2 一次完整累加从driver注册到task结果回传把一次累加计数的完整生命周期拆开看一共六步driver端创建并注册。你调用sc.longAccumulator(counter)或自己new一个累加器再sc.register(acc, name)时Spark会把累加器的初始状态注册到SparkContext里。注册的意义在于Spark需要追踪这个累加器便于后续在task结果中携带它的更新、在UI上展示、以及在某些重试场景下恢复初始状态。闭包序列化分发。当RDD算子引用了这个累加器对象Spark会像序列化普通变量一样把它连同闭包一起发给executor。每个executor拿到的又是一个副本。task执行前copy一份。更准确地说每个task在执行前会调用copy()方法把累加器复制一份独立的实例。task只在这个副本上执行add多个task互不干扰这也是累加器在并发环境下能保持线程安全的原因之一。task本地执行add。你在map或foreach里调用acc.add(1)实际上只更新了task自己的那个副本。task内可以调用任意多次add完全本地操作没有任何网络通信开销这就是累加器性能高的核心原因。task完成结果回传driver。task执行完毕后Spark会把累加器的本地副本值打包进task的TaskResult中随结果一起回传给driver。注意这里回传的不是累加过程中产生的每一个值而是经过本地合并后的一个快照。driver端执行merge。driver收到每一个task的结果后取出里面的累加器副本调用全局累加器的merge(other)方法把task的结果合并进全局状态。所有task的merge都完成之后你就能通过acc.value拿到统计结果了。整个流程用一句话概括task端只负责adddriver端负责merge中间靠task结果回传作为桥梁。2.3 内建累加器与老版本API的差异平时最常用的两个内建累加器是sc.longAccumulator和sc.doubleAccumulator分别对应LongAccumulator和DoubleAccumulator只支持数值累加。它们的好处是简单、快、不需要任何自定义代码。但很多老教程里还在用这样的写法val counter sc.accumulator(0, counter)这是Spark 1.x时代的API底层基于AccumulatorParam。如果你今天还在用Spark 2.0以上的版本建议直接换成sc.longAccumulator别再看老代码了。sc.accumulator在2.0之后已经标记为废弃虽然还能编译通过但类型体系、性能设计和UI展示都不如新版接口完善。另外提一嘴累加器和广播变量经常被放在一起对比广播变量是只读共享把大变量广播到executor供所有task读取累加器是只写共享task端只能写driver端只能读最终结果。两者一个负责下发放置数据、一个负责收集计算结果配合使用时可以覆盖很多分布式编程场景。3. 累加器的四个关键特点惰性触发、只写不读与重复计算很多文章介绍累加器特点时就写一句累加器是Spark提供的累加变量这远远不够。实操中真正决定你用不用、怎么用累加器的是下面这四个特点。前两个是表面特性后两个才是真正的深坑。3.1 特点一行动算子触发生效累加器是懒惰的这是Spark执行模型的一部分RDD上的transformation算子都是懒执行的累加器的更新也遵循这个规则。看这段代码val acc sc.longAccumulator(counter) val mapped rdd.map { x acc.add(1) x } // 只执行到这里不触发action println(acc.value) // 输出0map是transformation它只是构建了血统关系并没有真正跑起来。累加器的add一次都不会执行driver的acc.value自然还是0。只有当你调用count、collect、saveAsTextFile、foreach这类行动算子时job才真正开始执行add才会被调用。这个特性带来的一个陷阱是如果同一个RDD被多次行动算子触发且每次执行都会重新计算之前的transformation那累加器就会累加多次。比如val rdd someSource.map { x acc.add(1); x } rdd.count() // acc第一次累加 rdd.count() // acc再次累加值翻倍即使你用了rdd.cache()如果executor内存不够导致分区被提前清理或者cache之后源数据变了Spark仍然可能重算。因此在判断累加器数值对不对之前先确认这个RDD到底被行动算子执行了多少次。3.2 特点二executor端只写不读task里读到的永远不是全局值累加器在task里只能执行add你如果尝试在executor端读取acc.value读到的只是task本地副本的值通常是初始值或这个task自己刚刚累加的结果绝不是driver端累积了所有task之后的全量值。rdd.foreach { x acc.add(1) // 这里读到的value几乎可以确定是错误的、不完整的 if (acc.value 1000) { // 你永远等不到这个分支触发 } }所以在map或foreach里写当累加器达到某个阈值就做某件事这种逻辑是完全行不通的。原因也好理解如果每个task都要读取driver端的全局实时值那每次add都得做一次网络往返累加器就失去性能优势了而且多task并发执行时全局值时刻在变你读到某个值根本不能代表稳定的全局状态做控制流决策反而会得到错误结论。3.3 特点三task重试和stage重算会造成重复累加这是最大的坑这个是累加器在生产环境中最容易翻车的点没有之一。Spark的容错机制是通过重算来恢复。一个task因为节点宕机、executor失联、内存溢出等原因失败了TaskScheduler会把它重新调度到其他executor上执行。如果一个stage的shuffle输出文件丢失Spark会把这个stage及下游stage重新计算。在这些重试和重算的过程中累加器的add会被再次执行但是Spark不会自动把之前已经累加进去的数值回滚。举个例子一个task本来已经执行成功add了500次结果回传给driver了。但driver后来发现另一个stage需要重算这个task又被重新执行了一次又add了500次。最终driver上累加器的值就变成了1000而不是真实的500。这也是为什么Spark官方文档明确提醒对于需要在失败重试时保证结果精确的作业累加器不做这种保证。累加器在容错语义上是best-effort的它能满足监控、统计、诊断这类多算一点点可以接受的场景但不适合充当业务结果的计算工具。我遇到过一个实际案例某个数仓任务用累加器统计过滤掉的异常行数某天凌晨集群节点出问题一批task重试了3轮当天上报的异常行数直接翻了快3倍看起来像数据突然恶化了实际上只是重试叠加而已。3.4 特点四merge逻辑可自定义但设计时必须考虑交换性和结合性累加器merge(other)是用来合并两个累加器状态的。对于数值累加器merge就是加法天然满足交换律和结合律。但自定义累加器时你完全可以定义自己的合并逻辑比如并集、最大值、最值集合等。这里有一个隐含要求多个task的merge顺序是不确定的。driver可能先收到task A的结果再收到task B的结果也可能反过来执行merge时是逐个合并进全局累加器全局状态不保证按某个固定顺序合并。因此你的merge逻辑必须是可交换的、可结合的。比如取最大值求和取集合并集都满足但求平均值就不满足——A和B先合并再与C合并和B与C先合并再与A合并结果一样吗不一定。我见过有人想用累加器做平均值统计写出来的merge是(this.sum other.sum) / (this.count other.count)这就是典型的错误设计最终结果完全取决于merge顺序。遇到这类需求正确做法是累加器里只维护sum和count两个值value方法里最后再算平均值merge时只合并原始sum和count。3.5 特点小结一张表看清累加器的边界特点具体表现实操注意点惰性执行只有行动算子触发job时才更新同一RDD被执行多次累加器会重复累加只写不读task端只能add读.value得到的是本地副本不能在task里基于累加器值做控制流容错重算task重试、stage重算会重复执行add不适合做业务精确计量适合监控诊断自定义merge多个task结果的合并顺序不确定merge要满足交换律和结合律4. 自定义累加器实战用数据质量统计把原理跑通4.1 为什么需要自定义内置累加器的能力边界LongAccumulator和DoubleAccumulator只能做简单的数值累加。但真实场景里我们经常需要在一次作业运行过程中同时统计多种指标日志数据里某个字段的空值数量数值字段为负数的记录数非法枚举值的数量某几个字段的最大值、最小值过滤条件丢弃的各类原因计数如果每次都要单独定义一个LongAccumulator那会累积出一堆变量register代码和驱动端读取代码都变得很啰嗦。更合理的做法是把这些统计项封装进一个自定义累加器让add接收一条记录value返回一个完整的统计报告。4.2 完整代码一个数据质量累加器下面是我在数据清洗作业里用过的思路简化后分享出来。它实现的功能是统计输入日志中空值数、负数值、非法值数以及正常记录数。import org.apache.spark.util.AccumulatorV2 import scala.collection.mutable class DataQualityAccumulator extends AccumulatorV2[String, Map[String, Long]] { private val counters mutable.Map[String, Long]() override def isZero: Boolean counters.isEmpty override def copy(): DataQualityAccumulator { val acc new DataQualityAccumulator counters.foreach { case (k, v) acc.counters.put(k, v) } acc } override def reset(): Unit counters.clear() override def add(record: String): Unit { // 这里按实际解析结果决定统计哪个维度 record match { case uid_null counters(uid_null) counters.getOrElse(uid_null, 0L) 1L case negative_time counters(negative_time) counters.getOrElse(negative_time, 0L) 1L case invalid_city counters(invalid_city) counters.getOrElse(invalid_city, 0L) 1L case valid counters(valid) counters.getOrElse(valid, 0L) 1L case _ counters(other) counters.getOrElse(other, 0L) 1L } } override def merge(other: AccumulatorV2[String, Map[String, Long]]): Unit { other.value.foreach { case (k, v) counters(k) counters.getOrElse(k, 0L) v } } override def value: Map[String, Long] counters.toMap }使用方式也很简单val qualityAcc new DataQualityAccumulator sc.register(qualityAcc, data-quality) val df spark.read.json(/data/logs/2024-06-01) df.foreachPartition { iter iter.foreach { row val uid row.getAs[String](uid) val time row.getAs[String](event_time) val city row.getAs[String](city) if (uid null) qualityAcc.add(uid_null) else if (time null || time.toLong 0) qualityAcc.add(negative_time) else if (city null || city.isEmpty) qualityAcc.add(invalid_city) else qualityAcc.add(valid) } } val report qualityAcc.value println(s数据质量报告: $report)这里我用了foreachPartition而不是foreach目的很明确在每个分区内复用迭代逻辑减少add调用的外层开销。不过要注意无论foreach还是foreachPartition本质都是行动算子累加器都会正常生效。4.3 代码之外的四个注意点第一累加器类必须可序列化。它会被序列化后分发给executor所以内部不要持有不可序列化的对象比如没实现的连接池、非序列化的日志框架等。一个简单的检查方法是在类声明里继承SerializableAccumulatorV2已经继承了并且所有字段都用mutable.Map、Long这类可序列化类型。第二value方法要返回不可变快照。我上面返回的是counters.toMap而不是直接返回counters这个mutable.Map。如果直接返回可变Mapdriver端读完再修改很容易污染全局累加器状态引发并发问题。第三copy方法要真正复制状态。每个task执行前都会调用copy如果copy返回的是this本身或者共享同一个Maptask之间的add会互相污染最终统计结果会乱成一锅粥。copy里一定是new一个新对象把当前counters的键值逐个拷过去。第四严禁往累加器里塞明细数据。有些新手会想用自定义累加器收集所有非法行把原始数据都塞进累加器的集合里。这个想法很危险每个task都会把非法的原始数据回传driverdriver内存会被瞬时打爆。正确姿势是只统计计数原始明细写在executor本地或通过旁路日志输出绝不能让driver当数据汇集中心。4.4 为什么一定要sc.register可能有同学会问我不调sc.register累加器也能正常add、value啊register有什么用register至少有三层意义让Spark把累加器注册到SparkContext的累加器列表中driver可以随时拿到它的引用UI界面的Accumulators标签页里也会显示它的当前值方便你实时观察进度。注册是Spark恢复机制的一部分。在部分任务重试场景下Spark需要知道哪些累加器需要重置或重建注册过的累加器才能被正确管理。不注册的累加器在Web UI里没有名字、没有历史记录出了事你连值班日志都没得查。所以自定义累加器写完之后第一件事就是sc.register(acc, 可读名字)这个名字会显示在Spark UI里排查问题全靠它。5. 生产环境里的选型判断累加器、reduceByKey和Metrics谁更合适5.1 先看一张对比表方案精确性性能开销适用场景累加器不保证容错精确重试会重复累加很低task内纯本地操作作业内监控、数据质量统计、调试信息reduceByKey/aggregate精确失败重算后结果正确需要shuffle有网络IO业务最终聚合结果collect回driver计算精确但数据量大有OOM风险很慢driver成为瓶颈小数据量验证、调试Spark内建Metrics精确Spark自身统计无额外代码开销shuffle字节数、输入记录数等系统监控这个表的本质区别是什么累加器是尽力而为的统计工具reduceByKey才是保证正确的计算算子。两者的定位完全不同不要把累加器当成精确计数的替代品。5.2 我在生产环境里的使用习惯经验一累加器用来做过程监控不参与业务结果计算。比如在ETL作业里我会用累加器统计每条数据被过滤的原因等作业跑完这些统计写到日志或监控面板里用于判断数据质量波动。但如果这个统计结果要进入下游报表、参与金额计算我一定会用reduceByKey或agg函数重新计算绝不直接用累加器输出。经验二把累加器统计的记录总数和rdd.count()做交叉校验。这是个非常实用的技巧。比如累加器统计出valid invalid总共N条如果rdd.count()和N对不上说明要么累加器没有覆盖所有分支要么有stage重算导致重复累加。这种校验能帮你在一开始就发现代码逻辑或容错的隐患而不是等上线之后才暴露。经验三在DataFrame上使用累加器优先配合foreachPartition。DataFrame的foreach一行一行执行效率偏低foreachPartition一次处理一个分区可以同时结合add逻辑和本地缓存减少每行的调度开销。如果你的统计逻辑比较复杂比如要解析JSON、做正则匹配一个分区里再包一层迭代器循环能省不少事。经验四不要在生产环境里用累加器往外部系统实时推数据。累加器的执行时机是task粒度的task完成才会把值传给driver不是每add一次就通知一次。如果你想实时感知处理进度应该用Streaming或Structured Streaming里的状态更新机制而不是指望累加器。5.3 最后的排查技巧让Accumulators标签页帮你写作业写到最后分享一个我一直保留的习惯。每次跑Spark作业我都会在Web UI的Accumulators标签页里盯着自定义累加器的变化。这个页面会展示每个累加器的当前值和历史值比你在代码里println要直观得多。有一次线上作业的数据量级突然异常我第一时间打开Accumulators标签页发现某个累加器的值比上一批作业翻了正好一倍而其他指标都正常。顺着这个线索查下去发现是上游新加了一个重试机制导致同一份数据被计算了两次。如果没有这个UI页面单靠代码日志排查至少得多花一小时。还有一个经验如果哪天你的累加器数值看起来莫名其妙地大先怀疑stage重算或RDD被多次执行而不是怀疑自己的代码逻辑。这个排查顺序能帮你省下很多时间尤其是集群资源不稳定、任务频繁重试的时候。累加器本身不是什么高深机制但它是理解Spark分布式执行模型的一把钥匙。把task本地add、driver统一merge、重试会重复累加这几个核心点刻在脑子里用起来就不会踩大坑。至于什么时候该自己继承AccumulatorV2我的建议很简单当你需要统计的维度超过五六个、而且想在一次作业里集中汇总时就值得动手写了。