统计一批脏数据的条数这事在单机脚本里就是一行cnt 1但换成 PySpark 之后我第一次写完跑完发现计数器始终是 0排查了半天才反应过来transformation 是惰性的我没触发 action。后来又踩过一次更隐蔽的坑——同一个 RDD 算了两次计数直接翻倍。PySpark 累加器Accumulator就是为这类分布式只写变量准备的组件它能让你在 executor 端安全地往一个共享变量上加东西最后在 driver 端读总账。内置的 Long、Float、Collection 三类累加器能覆盖大部分计数场景但一旦你想按字段名分组统计、想收集限量样本、想维护一个 TopN 榜单就必须上自定义累加器。这篇内容我按自己的实操路径整理先讲清楚它和普通变量、广播变量的边界再把AccumulatorParam的两个方法拆开揉碎最后给一套能直接抄进 ETL 作业的数据质量看板实现以及一份出错时的排查速查表。刚上手 Spark 的同学可以照着代码跑通写过一段时间作业、被累加器坑过的同学可以重点看第三章和第五章。1. 先把定位说清楚累加器不是普通变量也不是广播变量1.1 分布式计数为什么会出错三个执行期的错位要理解累加器为什么必须存在得先在脑子里把 Spark 作业的执行过程切成三块driver 端的代码组织期、闭包序列化期、executor 端的 task 执行期。你在 driver 上写的cnt 0属于第一块它是一个 Python 对象住在 driver 进程的堆里。当你把它塞进rdd.map(lambda x: cnt 1)这种闭包时它会被序列化一份副本发到每个 executor每个 task 手里都有一份互相看不见的拷贝。task 里cnt 1算出来的结果既不会回传也不会合并任务结束就随进程一起被丢掉了所以 driver 上的cnt永远是 0。更麻烦的是一个 RDD 有多少个分区就有多少个 task每个 task 都有自己的副本。哪怕你换了写法把结果通过map输出成(1, ...)再去reduce求和你也绕开了这个问题但这意味着你要把中间结果真的变成数据在集群里搬运。对于我就想知道有多少条记录解析失败这种纯旁路统计把它混进主数据流里既浪费 shuffle又污染了业务逻辑。累加器解决的正是这个缺口它是一条从 executor 到 driver 的单向旁路通道。executor 端只能add不能读driver 端只能读value也可以 add但 driver 的 add 是直接改本地值不涉及回传。这条通道不参与血缘不产生 shuffle不改变 RDD 内容属于顺手记一笔的定位。这里有个概念一定要纠正很多人第一次看到共享变量这个词会以为 executors 之间能互相看到累加器的当前值。不是的。executor 端拿到的累加器是零值起步的本地副本它只负责本地累加任务结束时把增量回传给 driver由 driver 统一合并。也就是说累加器在语义上是多份局部和 driver 端合并而不是一块所有人都能读写的共享内存。想明白这一点后面很多诡异现象就都能解释了。1.2 累加器与广播变量的分工以及只写不读的约束Spark 里的共享变量就两个广播变量Broadcast和累加器Accumulator。它们的定位几乎是对称的我用一张表把关键差异列清楚选型的时候直接对照。维度广播变量 Broadcast累加器 Accumulator数据流向driver 到 executor只读executor 到 driver只写能否在 task 中读取可以就是它的用途不允许PySpark 会直接拦截报错能否在 task 中修改不允许本身不可变只能通过 add 累加传输方式每台 executor 一份走块管理器每个 task 结束回传增量适用场景维表、字典、模型参数计数、指标、脏数据样本典型误用用它存会变的中间状态用它当计数器控制流程分支累加器只写不读这条约束不是 Spark 故意为难你而是分布式语义决定的。假设你能在 task 里读累加器的当前值那么同一个 stage 里两个并行 task 读到的是同一个值还是各自的值如果读到 driver 的最新值那每次读都要跨网络同步性能崩掉如果读本地值那读到的是零值 本 task 到目前为止的累加量跟 driver 上的值对不上逻辑必然出错。所以 PySpark 选择了最干脆的做法在 task 里访问.value直接抛异常报错信息大意是Accumulator.value cannot be accessed inside tasks。你如果在本地模式调试时看到类似的报错不要去查 Spark 版本先检查是不是把.value写在了map、filter这类算子里。顺带说一句我的实际体会这条约束在写代码时是好事。它逼着你在 driver 上处理累加结果也就是逼着你的业务逻辑和监控逻辑分开——监控数据最后统一在 action 之后读取、打印或者上报。我见过太多把统计逻辑和业务分支搅在一起的代码全都是从在 map 里读一下累加器开始的。1.3 一个最小可运行示例五分 钟验证惰性与累积概念讲完直接上代码。本地模式四行就能跑建议你跟着敲一遍亲手感受一下累加器什么时候变、什么时候不变。下面这段代码我一直在用当作新同学入职的练手题。from pyspark.sql import SparkSession spark (SparkSession.builder .appName(accumulator-basic) .master(local[2]) .getOrCreate()) sc spark.sparkContext sc.setLogLevel(WARN) total sc.accumulator(0) bad sc.accumulator(0) rdd sc.parallelize(range(1000), 4) def check(x): total.add(1) if x % 7 0: bad.add(1) return x mapped rdd.map(check) print(before action:, total.value, bad.value) # 输出before action: 0 0 mapped.count() # 真正的计算发生在这一行 print(after 1st action:, total.value, bad.value) # 输出after 1st action: 1000 143 mapped.count() # 没缓存整个血缘从头重算 print(after 2nd action:, total.value, bad.value) # 输出after 2nd action: 2000 286这段代码藏着三个知识点。第一rdd.map(check)只是记录了血缘total.value仍然是 0这就是惰性求值的直观体现。第二第一次count()之后total是 1000、bad是 143因为 0 到 999 中能被 7 整除的数有 143 个floor(999/7)。第三第二次count()之后数字翻倍了——因为mapped没有被 cacheSpark 从源头重算了一遍check又被执行了一次累加器自然又加了一遍。提示这段翻倍的现象不是 bug是累加器的设计语义。写代码时请把累加器当成统计执行过程而不是统计数据本身的指标它天然是 at-least-once 的。如果你把mapped.cache()加在两次 action 之间第二次count()只会从缓存读check不再执行累加器就不会再涨。这一点在第五章排查数字偏大的时候非常关键先记住它。2. 内置累加器的三种用法与坑2.1 Long、Float 两种数值累加器与类型推断sc.accumulator()是创建累加器的入口不传第二个参数时PySpark 会根据初始值的 Python 类型自动选一个AccumulatorParam。传0会走LONG_ACCUMULATOR_PARAM传0.0走FLOAT_ACCUMULATOR_PARAM传复数走COMPLEX_ACCUMULATOR_PARAM。这个自动推断看起来贴心实际上有个坑计数场景一定用 int别用 0.0 起步。用浮点累加器做计数虽然结果通常还是整数但你会得到一个 float 类型的值跟别的整数指标拼在一起做 JSON 序列化的时候容易多出.0尾巴下游看板看着别扭。至于精度浮点累加器在百万级以上的累加会被浮点误差影响做金额统计时不要用它那种场景应该用DoubleAccumulator或者索引用自定义累加器配合 Decimal 处理。数值累加器最常见的写法就是sc.accumulator(0)然后在算子里acc.add(1)或者acc 1。这里有一个细节值得说只是add的语法糖实际调用的是同一个方法性能上没有区别。但在闭包里用acc 1有个隐患Python 会把它当成对闭包变量的重新赋值绑定某些老版本或者特殊写法下会报UnboundLocalError所以我个人的习惯是统一用acc.add(1)语义更明确也不容易出幺蛾子。还有一个容易被忽略的点累加器在 driver 上也是可以 add 的。你在 driver 代码里写total.add(100)它会直接改本地值不会产生任何跨进程通信。这在单元测试里有用——你可以在 driver 上预置一个基数再跑 action 观察增量是否符合预期。但千万别在 driver 里 add 完之后又指望 executor 的更新能盖掉它两者是相加关系不是覆盖关系。2.2 CollectionAccumulator 能不用就别用Spark 3.0 开始PySpark 提供了sc.collectionAccumulator()底层是 JVM 侧的CollectionAccumulator用法很直白executor 里samples.add(row)driver 端samples.value拿到一个 Java List。很多人第一次看到它就想拿它做收集所有异常数据我劝你先冷静一下。这个累加器的本质是把数据全都堆在 driver 内存里。它的合并操作就是往一个列表里 append不设上限。你想想一个 TB 级作业里有 0.1% 的脏数据那也是几十万行如果这些行里每一行都是几百字节的原始记录driver 直接 OOM。我在一个日志清洗作业里就见过这个事故开发同学用collectionAccumulator收集解析失败的原始行做排查测试数据只有几千条所以没问题上生产之后 driver 内存从 4G 撑到挂掉整个作业重启排查了半天才定位到是累加器吃掉了内存。正确的用法是限量采样。要么在 add 之前自己判断长度要么干脆上自定义累加器加上限控制第三章会给出完整实现。如果只是想在 UI 上看到几条样本take(5)配合打印日志就够了不一定非要用累加器。真要用请记住三条只在排查期开、样本量控制在两位数、用完及时从代码里摘掉。还有一个细节collectionAccumulator的.value返回的是 Java 集合对象Python 侧遍历它要经过 py4j 转换list(samples.value)[:5]这种写法在样本量大的时候会有明显的性能损耗这也是催你限量的另一个理由。注意collectionAccumulator在 PySpark 3.0 之前不可用。如果你的集群还在 2.x得用自定义AccumulatorParam加一个 list 来实现写法在第三章。2.3 action 内更新与 transformation 内更新的语义差异这一节是全文最重要的技术点之一建议反复读两遍。Spark 官方对累加器的保证是这样描述的如果累加器的更新只发生在 action 内部例如foreach、foreachPartition那么每个 task 的更新只会被应用一次如果更新发生在 transformation 内部例如map、filter、flatMap那么当某个 task 因为失败重试、或者因为血缘重算而再次执行时更新会被重复应用。这句话翻译成实操规则就是两条。第一纯计数场景优先把 add 放在 action 型算子里。比如你要统计解析失败的行数与其在map里顺手加一下不如专门写一个foreachPartition把所有校验逻辑集中起来。这样每个 task 只执行一次即使前面有 shuffle 导致的 stage 重算也不会重复加。代价是多扫一遍数据——但如果你的数据来自 shuffle 之后的结果或者已经 cache 过这个代价是可以接受的。第二如果你必须在 transformation 里加那就要接受值可能偏大的事实并且用 cache 来降低重复执行的概率。注意是降低不是消除。因为 task 失败重试是集群的常态executor 被 kill、内存溢出、推测执行触发重复计算这些都会导致同一个 task 被跑两遍。你唯一能做的是让累加器统计执行次数而不是数据条数——如果你需要的是精确的数据条数那就老老实实用df.count()或者聚合算子别用累加器。我在实际项目里总结出一条分界线累加器用于监控和调试聚合算子用于业务口径。凡是会写进报表、会跟财务对账的数字一律走groupBy().agg()把结果落到一张表里凡是这批数据里有多少条解析异常有多少个分区是空分区这种运维性质的数字才用累加器。这条界线划清楚之后团队里关于累加器数字对不上的争论少了一大半。3. 自定义累加器AccumulatorParam 的两个方法到底怎么写3.1 为什么 PySpark 让你实现的是 AccumulatorParam 而不是 AccumulatorV2如果你搜过Spark 自定义累加器大概率会看到一堆讲AccumulatorV2的文章列出isZero、copy、reset、add、merge、value这些方法然后让你照抄。你在 PySpark 里试着class MyAcc(AccumulatorV2)会发现根本跑不通。原因在于AccumulatorV2是Scala/Java 侧的抽象类它属于 JVM 世界。PySpark 并没有把它的子类化能力暴露给 Python——Python 对象没法直接继承 JVM 类并让 JVM 侧的调度器调用它的方法。所以 PySpark 给了一套 Python 侧的接口AccumulatorParam放在pyspark.accumulators模块里只需要实现两个方法zero(initialValue)和addInPlace(value1, value2)。两者不是没关系而是 Spark 帮你做了一层适配Python 侧的自定义累加器最终会被包装注册到 JVM 上JVM 在合并各个 task 的回传结果时会回调到 Python 侧的addInPlace来完成合并。所以理解AccumulatorParam最快的方式是把它对应到AccumulatorV2的几个方法上去AccumulatorV2 方法JVM 侧Python 侧对应物作用zero()zero(initialValue)生成一个零值对象供每个 task 起步用add(v)addInPlace(current, delta)把增量加进来merge(other)addInPlace(current, delta)合并另一个同类型累加器value直接访问acc.value取最终结果isZero/copy/reset无需实现由框架在 JVM 侧统一处理看明白这张表你就知道Python 侧把局部累加和跨节点合并两件事合并成了同一个方法addInPlace。这也解释了为什么它必须满足交换律和结合律合并顺序是不确定的driver 合并 task 结果的顺序取决于哪个 task 先结束如果你在addInPlace里写了谁先来谁是基准这类有顺序依赖的逻辑结果会随调度而变。这一点是自定义累加器最容易翻车的地方。那什么时候该用 JVM 侧的自定义AccumulatorV2只有一种情况你的累加器逻辑需要用 Java/Scala 的高性能数据结构实现然后再暴露给 PySpark 调用。这种混合编程的维护成本很高除非累加器对象特别大、更新特别频繁否则不值得。绝大多数 PySpark 作业老老实实写AccumulatorParam就够了。3.2 zero 与 addInPlace 的契约三条约定的红线写自定义累加器的时候有三个约束必须守住我把它叫三条红线。第一条zero()每次都要返回一个全新的对象不能返回同一个全局变量。这个坑非常隐蔽。如果你这么写# 错误示范千万别这么写 _SHARED [] class BadParam(AccumulatorParam): def zero(self, initial_value): return _SHARED # 所有 task 共用同一个 list def addInPlace(self, v1, v2): v1.extend(v2) return v1zero()返回的是模块级的同一个列表对象而同一个 Python worker 进程里会连续跑多个 tasktask A 往这个列表里塞的数据task B 一进来就已经看到了。最终结果会莫名其妙地偏大而且偏大的幅度跟 worker 复用次数有关完全不可复现。正确写法永远是return []每次构造新对象。第二条addInPlace要原地修改第一个参数并返回它。PySpark 的历史实现里某些路径依赖原地修改这个约定虽然Accumulator.add会把返回值赋回去理论上返回新对象也没问题但为了兼容性和可读性我建议统一按改完value1再return value1来写。如果你确实需要返回新对象比如 TopN 场景下排序后截断那也没关系只要保证返回值类型一致就行。这里的关键是不能返回None这是新手常犯的错——只写了v1[k] ...忘了 return结果累加器值变成None后面再 add 就报AttributeError。第三条整个AccumulatorParam实例必须能被 pickle。累加器对象和它的 param 都要被序列化发到 executor走的是 Python 的 pickle 通道。如果你的 param 里持有了数据库连接、SparkSession、lambda 闭包这些不可序列化的东西任务提交时会直接抛PicklingError。我踩过的具体一次是param 里存了一个 logger 对象想用来记录日志结果整个作业起不来。正确做法是只保留纯数据结构int、str、tuple、简单 dict需要日志就在 executor 里用logging现取一个 logger。在这三条之上还有一条性能建议addInPlace会被每条记录调用一次如果你在map里逐条 add 的话。所以这个方法里不要做深拷贝、不要做格式化、不要写文件、不要打印。我见过有人在里面写json.dumps做调试QPS 直接掉到原来的十分之一。3.3 三个可以直接抄的自定义累加器下面三个实现是我在实际作业里反复用到的按复杂度递增排列你可以直接抄。第一个字典计数器用于按字段名分组统计。这个解决的是字段级指标的问题——同一个校验流程里要同时统计age_null、amount_negative、email_invalid等一堆互不相干的计数用多个累加器太啰嗦用 dict 一把梭。from pyspark import AccumulatorParam class DictCounterParam(AccumulatorParam): 把 (指标名 - 计数) 的字典逐分区合并 def zero(self, initial_value): return {} def addInPlace(self, v1, v2): if not v2: return v1 for key, cnt in v2.items(): v1[key] v1.get(key, 0) cnt return v1用法是在 driver 上field_err sc.accumulator({}, DictCounterParam())然后在 executor 里构造一个局部 dict一个分区只 add 一次。注意addInPlace里的if not v2判断不只是防御性编程它是真的能省时间——空分区在真实数据里很常见尤其是过滤之后的分区。第二个限量样本收集器替代collectionAccumulator。核心是在合并阶段就做截断保证 driver 侧的内存占用有硬上限。class SampleParam(AccumulatorParam): 最多保留 limit 条样本防止 driver OOM def __init__(self, limit20): self.limit limit def zero(self, initial_value): return [] def addInPlace(self, v1, v2): if len(v1) self.limit or not v2: return v1 for item in v2: if len(v1) self.limit: break v1.append(item) return v1这里的limit存在 param 实例上会跟着 pickle 一起分发所以每个 executor 都知道上限。有个细节值得注意因为这个实现是先到先得最终保留的样本取决于哪些分区先完成合并不是随机的。如果你需要随机采样就在 executor 侧用random.random() 0.001做一层前置筛选再交给累加器。我有次排查一个只在夜间批次出现的数据问题加了这个累加器之后一次作业就把出问题的 20 条样本抓出来了比翻几 TB 的日志快得多。第三个TopN 累加器用于找极值。典型场景是找出处理耗时最长的 10 个文件或者找出金额最大的 10 笔异常交易。class TopNParam(AccumulatorParam): 维护当前最大的 n 个 (score, label) 元组 def __init__(self, n10): self.n n def zero(self, initial_value): return [] def addInPlace(self, v1, v2): if not v2: return v1 merged v1 v2 merged.sort(keylambda t: t[0], reverseTrue) return merged[: self.n]这个实现里addInPlace返回的是新列表不是原地修改v1。前面说过这在新版 PySpark 上是安全的但为了保险你也可以改成v1[:] merged[: self.n]后再return v1两种写法我都跑过结果一致。TopN 的性能关键是n别开太大因为每个 task 回传的数据量至少是n条n10和n10000对网络和合并的开销完全是两个量级。提示所有自定义累加器都要在 driver 上创建然后通过闭包引用传给 executor。不要在 executor 里 new 一个累加器那样它不会注册到 driver你永远读不到值。4. 把累加器接进真实 ETL一个数据质量看板的完整落地4.1 需求拆解与指标设计光有零件不行得拼成整机。我拿一个真实的订单清洗作业举例需求是这样的从原始订单表读数据做字段校验和类型转换坏数据要过滤掉但不能静默丢弃运营同学需要一个这批数据质量怎么样的反馈。按照这个需求我把指标分成四类这个分类方法你可以直接套用到自己的项目里。第一类是总量指标输入总条数、输出总条数、过滤条数。第二类是字段级异常指标每个字段各自有多少条为空、多少条格式不合法、多少条超出取值范围。第三类是样本指标每类异常保留最多 5 条原始记录方便运营同学拿去向业务方反馈。第四类是分布指标比如订单金额被分到 10 个区间里各自有多少条用来快速看数据分布有没有异常。这四类指标刚好对应三章里的三个自定义累加器加一个数值累加器结构非常清晰。设计的时候有几个取舍点值得说一下。第一为什么不用 DataFrame 的聚合算子算这些指标因为字段校验逻辑是行级的、带业务分支的用 SQL 表达式写出来会是一坨巨大的when().otherwise()嵌套可读性差且不好维护而用累加器可以在 Python 函数里写正常的 if-else逻辑一目了然。第二为什么样本要限量因为样本是用来给人看的5 条和 5000 条对排查的价值差别不大但对 driver 内存的影响天差地别。第三这些指标为什么不写回 Hive 表可以写但那样会多一次写操作和一次调度依赖对于每次跑完看一眼的监控需求直接打印成 JSON 打日志、让日志采集系统去解析成本最低。4.2 代码分层实现解析、校验、汇总代码我按三层来组织driver 层的累加器定义、executor 层的校验逻辑、收尾层的指标输出。这样分层的好处是累加器的定义集中在一处方便审查和复用。import json from pyspark import AccumulatorParam class DictCounterParam(AccumulatorParam): def zero(self, initial_value): return {} def addInPlace(self, v1, v2): if not v2: return v1 for k, n in v2.items(): v1[k] v1.get(k, 0) n return v1 class SampleParam(AccumulatorParam): def __init__(self, limit5): self.limit limit def zero(self, initial_value): return [] def addInPlace(self, v1, v2): if len(v1) self.limit or not v2: return v1 for item in v2: if len(v1) self.limit: break v1.append(item) return v1 class BucketParam(AccumulatorParam): 金额分布buckets 是上界列表例如 [0, 100, 1000, 10000] def __init__(self, buckets): self.buckets buckets def zero(self, initial_value): return [0] * (len(self.buckets) 1) def addInPlace(self, v1, v2): for i, n in enumerate(v2): v1[i] n return v1driver 层创建累加器注意BucketParam的初始值要和zero返回的形状一致长度是len(buckets) 1field_err sc.accumulator({}, DictCounterParam()) bad_samples sc.accumulator([], SampleParam(limit5)) amount_dist sc.accumulator([0] * 5, BucketParam([0, 100, 1000, 10000])) total_in sc.accumulator(0) total_out sc.accumulator(0)executor 层是整个逻辑的核心。关键技巧是把累加操作放在foreachPartition里并且在分区内部先做本地聚合这样每个分区只跟 driver 通信一次而不是每条记录一次def scan_partition(rows): local_err {} local_samples [] local_buckets [0] * 5 n_in 0 n_out 0 for row in rows: n_in 1 d row.asDict() keep True if d.get(order_id) is None: local_err[order_id_null] local_err.get(order_id_null, 0) 1 keep False age d.get(age) if age is None: local_err[age_null] local_err.get(age_null, 0) 1 elif age 0 or age 120: local_err[age_out_of_range] local_err.get(age_out_of_range, 0) 1 keep False amount d.get(amount) if amount is None: local_err[amount_null] local_err.get(amount_null, 0) 1 keep False elif amount 0: local_err[amount_negative] local_err.get(amount_negative, 0) 1 keep False else: if amount 100: local_buckets[0] 1 elif amount 1000: local_buckets[1] 1 elif amount 10000: local_buckets[2] 1 elif amount 100000: local_buckets[3] 1 else: local_buckets[4] 1 if not keep and len(local_samples) 5: local_samples.append({ order_id: d.get(order_id), age: age, amount: amount, }) if keep: n_out 1 # 一个分区只跟 driver 通信这几次 total_in.add(n_in) total_out.add(n_out) field_err.add(local_err) amount_dist.add(local_buckets) if local_samples: bad_samples.add(local_samples)这里有几个我的实操习惯。第一row.asDict()只在确实需要多字段的时候调用如果只需要一两个字段直接row.xxx更快因为asDict会构造整个字典。第二局部变量用local_前缀一眼就能和数据本身区分开。第三local_samples在分区内也限制了 5 条避免一个极端脏的分区造出几千条样本再交给累加器截断——虽然累加器会截断但分区内存先炸了。收尾层在 action 之后执行把结果打成一行 JSON方便日志系统采集df spark.read.parquet(s3://bucket/order_raw/) df.foreachPartition(scan_partition) report { total_in: total_in.value, total_out: total_out.value, filtered: total_in.value - total_out.value, field_errors: field_err.value, amount_distribution: dict(zip( [100, 100-1k, 1k-10k, 10k-100k, 100k], amount_dist.value)), samples: bad_samples.value, } print(DQ_REPORT json.dumps(report, ensure_asciiFalse))foreachPartition是 action所以这次统计的语义是每个分区恰好执行一次前面讨论的重复计数问题在这里不存在。这也是我强烈建议把校验逻辑放进foreachPartition的原因——同样的逻辑如果写在map里再配合count()你会因为各种重算而对不上账。当然代价是数据只走了一遍、没有产出清洗后的结果。如果作业本身还要输出干净数据那就把校验函数抽出来在map里过滤、在foreachPartition里统计两条血缘分开跑各自语义清晰。4.3 指标怎么用Spark UI、日志与告警的衔接指标算出来只是第一步怎么让它产生价值才是关键。我的做法是三层。第一层是Spark UI。累加器会自动出现在 Stage 页面里每个 stage 下面有一块累加器区域能看到这个 stage 里累加器的更新量。这对于定位哪个 stage 在产出脏数据特别有用。比如你发现总量对不上打开 UI 一看是某个 stage 的累加器更新量异常大那基本可以确定是这个 stage 的分区里数据有问题。需要提醒的是Python 侧自定义累加器在 UI 上的展示信息不如 JVM 侧丰富通常只能看到 ID 和数值没有友好的名字所以我一般会配合日志一起看。第二层是日志。上面那个DQ_REPORT前缀就是给日志采集用的采集系统按前缀抓一行 JSON 解析入库。这种方式的成本极低不需要任何额外的表结构而且天然带上了每次作业的执行时间戳。我在多个项目里都用了这套运营同学直接在 BI 里看趋势图数据质量突然劣化的时候曲线会肉眼可见地翘起来。第三层是主动告警。在收尾的 driver 代码里加一个简单判断比如过滤率超过 5% 就抛异常让作业失败或者调用内部的告警接口。这里要克制不要什么指标都告警否则很快就会被忽略。我的经验是只对两类指标告警过滤率和空分区数。过滤率突增说明上游数据出了变化空分区数突增说明上游的分区策略变了或者数据源断流了这两个都是需要人立刻介入的信号。至于字段级的细分指标看板上有就行不用告警。提示不要用累加器的值做业务逻辑分支判断比如如果失败数超过阈值就跳过后续处理。因为累加器只有在 action 完成后才可读等你读到的时候数据已经处理完了来不及做任何分支。5. 常见问题速查与性能调优5.1 值不对的六种典型场景对照排查累加器出的问题90% 集中在值不对上。我把这些年遇到的场景整理成一张表出问题的时候从上往下对照就行。现象根本原因快速验证方式处理方式始终为 0只写了 transformation 没触发 action在代码里补一个count()补 action或改写成foreachPartition始终为 0闭包里引用的是 driver 上的另一个同名变量打印id(acc)对比检查闭包捕获的变量名是否被局部变量遮蔽恰好翻倍同一血缘被触发两次RDD 未缓存看 DAG 里是否有重复 stage加cache()或把统计挪到 action 里随机偏大task 失败重试或推测执行导致重复执行看 UI 里的 task 重试次数把 add 移到 action 内接受不精确性比预期小部分 task 被 kill回传的增量丢失看 UI 里的 failed task 数减少单 task 数据量避免 OOM 被杀值类型不对addInPlace忘了 return或返回类型不一致打印type(acc.value)检查所有分支都有 return返回类型统一第一条和第三条是最常见的。我印象最深的一次是帮同事排查计数翻倍代码看起来完全正常最后发现是他在map里做统计然后分别用了df.count()和df.show()两个 action 来调试两个 action 各触发了一次全量计算。这种错误在本地小数据上完全看不出来只有上量之后才会发现数字不对。还有一个特别隐蔽的场景值得单独说DataFrame 的优化器可能让你的统计代码根本不执行。比如你写了df.filter(...).map(...)之后再统计如果 Catalyst 判断某些列用不上会做列裁剪某些情况下整个算子的执行顺序会被调整导致你以为会执行的map实际没跑。这不是累加器的锅是 Catalyst 的优化行为。规避方法很简单统计逻辑尽量放在foreachPartition或者rdd.mapPartitions这种不容易被优化的地方不要指望 DataFrame 的算子会老老实实按你写的顺序执行。5.2 性能上的三个抓手累加器本身的开销不大但用错了会变成瓶颈。三个抓手按收益排序。第一减少 add 的调用次数。这是收益最大的一条。前面例子里的做法是在foreachPartition里先做本地聚合一个分区只 add 一次。对比一下两种写法的开销逐条 add 意味着每条记录都要进一次 Python 方法调用、做一次 dict 更新分区级 add 意味着每 128MB 数据一个分区的典型大小只调用几次。差了三四个数量级。对于自定义累加器尤其明显因为addInPlace里还可能有遍历和合并逻辑。第二控制累加器对象的大小。累加器的值在每个 task 结束时都要序列化回传对象越大序列化和网络开销越大。字典计数器要注意别把基数很高的字段比如用户 ID放进 key 里那样字典会膨胀到几百万个键回传的时候直接卡住。同理TopN 的n别开大样本列表的limit别开大。我的经验值是单个累加器的回传体积控制在几十 KB 以内超过 1MB 就要重新审视设计。第三注意 driver 侧的内存。所有累加器的最终值都堆在 driver 内存里collectionAccumulator和没有限量设计的样本收集器是重灾区。除了限流还可以在收尾代码里读完值之后主动del掉不再需要的引用虽然 Python 的垃圾回收不一定立刻生效但至少不会因为闭包持有而一直存活。还有一个跟性能相关但经常被忽略的点累加器的数量不要太多。我见过一个作业定义了三十多个累加器每个字段一个理由是这样看着清晰。结果是每个 task 结束时要回传三十多份增量driver 要处理三十多次合并而且 Spark UI 上的累加器列表长到没法看。正确做法是像我第三章那个字典计数器一样用一个累加器承载一组同构的指标用 key 区分。数量控制在五到八个以内比较舒服。5.3 我在实际项目里踩过的几个坑最后分享几个不在官方文档里、但确实花了我不少时间的问题。坑一PySpark 的累加器没有 reset 方法。Scala 的AccumulatorV2有reset()但 PySpark 的Accumulator没有暴露这个能力。如果你在同一个 driver 里想复用累加器、把值清零重算是做不到的——只能重新sc.accumulator(...)创建一个新的。这在写循环式的批量处理代码时会咬你一口。我的应对方式是把累加器的创建封装成一个函数每次循环调用一次函数拿到全新的累加器。坑二在 Jupyter 里反复跑同一个 cell值会累积。因为 SparkSession 是复用的累加器对象也是同一个。你跑第二次的时候数值翻倍很容易误以为是代码逻辑错了。我的习惯是每次重跑都把累加器创建和 action 放在同一个 cell 里跑之前先Restart Kernel这样能排除掉状态污染。坑三acc.value返回的是内部对象的引用不是拷贝。对于 list 和 dict 类型的自定义累加器你在 driver 上拿到acc.value之后如果直接pop()或者append()会改掉累加器的内部状态。虽然通常不影响后续逻辑但会让日志输出和实际值不一致排查问题时非常误导。安全的做法是读取后立刻做一次浅拷贝samples list(bad_samples.value)。坑四Streaming 场景下的重复计数。Structured Streaming 的foreachBatch里用累加器作业从 checkpoint 重启之后会重新处理一批数据累加器的值不会从 checkpoint 恢复它不在状态管理范围内所以你会看到指标在重启后从 0 开始。这个行为不是 bug但如果你把累加器的值写进了下游的监控表就会出现数据倒挂的曲线。我的处理方式是在 Streaming 场景里干脆不用累加器改用foreachBatch里直接对 batch DataFrame 做聚合然后写入结果表这样天然具备幂等性。坑五本地模式和集群模式的行为差异。本地模式下master(local[1])只有一个 core所有 task 串行执行你不会遇到并发合并的问题换成local[4]或者上集群之后合并顺序不确定的问题才会暴露出来。所以自定义累加器写完一定要用local[4]跑一遍再上集群。这个习惯帮我提前发现了至少两次addInPlace的交换律问题——本地单线程跑的时候数字永远是对的多线程一跑就偏。上面这些坑归根结底都指向同一个认知累加器是一个弱一致性的旁路统计工具它的定位是给你提供快速的、近似的、用于观察的反馈而不是精确的业务口径。把它用在对的位置它是排查分布式作业最趁手的工具之一一次作业就能告诉你数据在哪个环节出了问题、比例是多少、样本长什么样用错了位置它会变成一个永远对不上账的黑洞让你在反复的调试里消耗掉大量时间。我个人在写任何一个稍微复杂的 ETL 时都会先花十分钟把累加器埋好跑第一遍就拿到数据质量画像后面的调优才有方向。