企业官网建设流程全解析

先说一个几乎所有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(s"task内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结果回传

把一次累加计数的完整生命周期拆开看，一共六步：

1. driver端创建并注册。你调用sc.longAccumulator("counter")或自己new一个累加器再sc.register(acc, "name")时，Spark会把累加器的初始状态注册到SparkContext里。注册的意义在于，Spark需要追踪这个累加器，便于后续在task结果中携带它的更新、在UI上展示、以及在某些重试场景下恢复初始状态。

2. 闭包序列化分发。当RDD算子引用了这个累加器对象，Spark会像序列化普通变量一样，把它连同闭包一起发给executor。每个executor拿到的又是一个副本。

3. task执行前copy一份。更准确地说，每个task在执行前会调用copy()方法，把累加器复制一份独立的实例。task只在这个副本上执行add，多个task互不干扰，这也是累加器在并发环境下能保持线程安全的原因之一。

4. task本地执行add。你在map或foreach里调用acc.add(1)，实际上只更新了task自己的那个副本。task内可以调用任意多次add，完全本地操作，没有任何网络通信开销，这就是累加器性能高的核心原因。

5. task完成，结果回传driver。task执行完毕后，Spark会把累加器的本地副本值打包进task的TaskResult中，随结果一起回传给driver。注意，这里回传的不是"累加过程中产生的每一个值"，而是经过本地合并后的一个快照。

6. driver端执行merge。driver收到每一个task的结果后，取出里面的累加器副本，调用全局累加器的merge(other)方法，把task的结果合并进全局状态。所有task的merge都完成之后，你就能通过acc.value拿到统计结果了。

整个流程用一句话概括：task端只负责add，driver端负责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) // 输出0

map是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 特点小结：一张表看清累加器的边界

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，所以内部不要持有不可序列化的对象，比如没实现的连接池、非序列化的日志框架等。一个简单的检查方法是：在类声明里继承Serializable（AccumulatorV2已经继承了），并且所有字段都用mutable.Map、Long这类可序列化类型。

第二，value方法要返回不可变快照。我上面返回的是counters.toMap，而不是直接返回counters这个mutable.Map。如果直接返回可变Map，driver端读完再修改，很容易污染全局累加器状态，引发并发问题。

第三，copy方法要真正复制状态。每个task执行前都会调用copy，如果copy返回的是this本身或者共享同一个Map，task之间的add会互相污染，最终统计结果会乱成一锅粥。copy里一定是new一个新对象，把当前counters的键值逐个拷过去。

第四，严禁往累加器里塞明细数据。有些新手会想用自定义累加器收集"所有非法行"，把原始数据都塞进累加器的集合里。这个想法很危险：每个task都会把非法的原始数据回传driver，driver内存会被瞬时打爆。正确姿势是只统计计数，原始明细写在executor本地或通过旁路日志输出，绝不能让driver当数据汇集中心。

4.4 为什么一定要sc.register

可能有同学会问：我不调sc.register，累加器也能正常add、value啊，register有什么用？

register至少有三层意义：

1. 让Spark把累加器注册到SparkContext的累加器列表中，driver可以随时拿到它的引用，UI界面的Accumulators标签页里也会显示它的当前值，方便你实时观察进度。
2. 注册是Spark恢复机制的一部分。在部分任务重试场景下，Spark需要知道哪些累加器需要重置或重建，注册过的累加器才能被正确管理。
3. 不注册的累加器，在Web UI里没有名字、没有历史记录，出了事你连值班日志都没得查。

所以，自定义累加器写完之后，第一件事就是sc.register(acc, "可读名字")，这个名字会显示在Spark UI里，排查问题全靠它。

5. 生产环境里的选型判断：累加器、reduceByKey和Metrics谁更合适

5.1 先看一张对比表

这个表的本质区别是什么？累加器是"尽力而为"的统计工具，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，我的建议很简单：当你需要统计的维度超过五六个、而且想在一次作业里集中汇总时，就值得动手写了。