Spark Scala中Accumulator配合filter、withColumn使用时计数值错误

计数翻倍本质是两个机制共同导致的：

• Spark转换算子是惰性执行的，只有遇到action算子才会真正触发计算，而且如果中间结果没做缓存，每触发一个action都会从头重算整条DAG链路
• 累加器的更新逻辑如果写在map/filter这类转换算子中，只要算子被重跑（不管是任务重试、还是不同action触发的链路重算），累加器就会重复累加

你的代码里一共触发了两个action：第一个是dataframe.show()，第二个是后面取文件名的collectAsList()。因为你没有缓存filter后的中间结果，第二个action触发时会重新跑一次读文件、filter的全流程，累加器就会再累加一遍数值，原本2条有效、3条无效的数据，跑两次自然就变成4条有效、6条无效。
你删掉withColumn和后续文件名计算逻辑之后，代码里只剩show()一个action，整条链路只计算一次，累加器结果自然就对了。
至于你在filter里打印的计数看着正常，是因为那些打印是executor端单task运行时的本地值，不是driver端最终汇总的全量值，参考意义不大。

两种方案都能解决问题，优先选第二种，生产环境稳定性更高：

方案1：最小改动，加缓存避免重算 #

直接在filter逻辑之后加cache()，把过滤后的结果持久化，后续所有action都复用缓存的结果，不会重跑filter逻辑，累加器就不会重复计数，代码调整点很少：

val badDataCountAcc = spark.sparkContext.longAccumulator("BadDataAcc")
val goodDataCountAcc = spark.sparkContext.longAccumulator("GoodDataAcc")

val dataframe = spark
      .read
      .format("csv")
      .option("header", true)
      .option("inferSchema", true)
      .load(path)
      .filter(data => {
        val matcher = regex.matcher(data.toString())
        if (matcher.find()) {
          goodDataCountAcc.add(1)
          true
        } else {
          badDataCountAcc.add(1)
          false
        }
      })
      .cache() // 核心改动：缓存filter后的中间结果，避免后续action触发重算
      .withColumn("FileName", input_file_name())

dataframe.show()
val filename = dataframe.select("FileName").distinct()
val name = filename.collectAsList().get(0).toString()

println("Bad data Count Acc: " + badDataCountAcc.value)
println("Good data Count Acc: " + goodDataCountAcc.value)

dataframe.unpersist() // 计算完成后释放缓存资源

提醒：把filter里的println删掉，分布式任务中executor的打印不会统一收集到driver端，打了也看不到全量日志，还会拖慢计算速度

方案2：弃用自定义累加器，用原生DataFrame API统计（生产环境首选） #

在转换算子中自定义维护累加器本身就是Spark不推荐的写法，很容易因为重算、任务重试出现计数不准的问题。直接用内置函数打标记、聚合统计，不管触发多少次action都不会出计数错误：

import org.apache.spark.sql.functions._

val processedDf = spark
      .read
      .format("csv")
      .option("header", true)
      .option("inferSchema", true)
      .load(path)
      // 打是否为有效数据的标记，注意把your_target_column换成你实际做正则匹配的字段名
      .withColumn("is_valid", when(col("your_target_column").rlike(regex), lit(true)).otherwise(lit(false)))
      .withColumn("FileName", input_file_name())
      .cache()

// 直接聚合统计有效、无效数据量
val countRes = processedDf.groupBy("is_valid").count().collect()
val countMap = countRes.map(row => row.getAs[Boolean]("is_valid") -> row.getAs[Long]("count")).toMap
val goodCnt = countMap.getOrElse(true, 0L)
val badCnt = countMap.getOrElse(false, 0L)

// 取有效数据（和原逻辑filter后保留的数据完全一致）
val validData = processedDf.filter(col("is_valid")).drop("is_valid")
validData.show()

// 取文件名
val fileName = processedDf.select("FileName").distinct().head().getString(0)

println(s"有效数据条数：$goodCnt")
println(s"无效数据条数：$badCnt")
println(s"输入文件名：$fileName")

processedDf.unpersist()

这个方案没有自定义累加器的坑，所有计算都是Spark原生优化过的，性能和稳定性都更好。

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