Scala Seq.reduceLeft未更新累加器值？DataFrame合并异常求助

你的问题出在可变BloomFilter的原地修改（mergeInPlace）导致的闭包副作用，结合Spark的惰性求值机制，最终破坏了DataFrame的依赖链：

1. 你在reduceLeft中使用mergeInPlace修改了tupNewer._2这个可变BloomFilter对象，而之前创建的UDF（newerDfBloomFilterUDF）持有该对象的引用。
2. Spark的DataFrame是惰性求值的，所有转换操作不会立即执行，直到调用show()、collect()等触发计算的方法。
3. 当最终执行combined._1.show()时，所有之前的UDF都会重新计算，此时它们引用的BloomFilter已经被多次mergeInPlace修改，导致过滤逻辑完全偏离了当时的预期：
   • 第一次迭代中原本应该保留的c，因为BF已经合并了后续元素，会被错误过滤；
   • 后续迭代的过滤逻辑也会因为BF的修改而失效，最终只剩下最初的b记录。

避免使用原地修改的mergeInPlace，而是每次创建新的BloomFilter实例进行合并，确保每个UDF引用的都是独立的、对应迭代状态的BF：

示例修正代码（以Guava BloomFilter为例） #

println("print input dfs")
dfList.foreach(_._1.show(20, false))
val combined = dfList.reverse.reduceLeft((tupNewer, tupOlder) => {
   println("tupNewer")
   tupNewer._1.show(20, false)
   println("tupOlder")
   tupOlder._1.show(20, false)
   val newerDfBloomFilterUDF = udf((s: String) => !tupNewer._2.mightContain(s))
   val filteredOlderDf = tupOlder._1.filter(newerDfBloomFilterUDF(col("id")))
   val unionedDf = tupNewer._1.union(filteredOlderDf)
   println("unionedDf")
   unionedDf.show(20, false)
   // 创建新的BloomFilter实例，合并新旧BF的内容
   val mergedBloomFilter = BloomFilter.create(tupNewer._2.hashFunction(), tupNewer._2.expectedInsertions())
   tupNewer._2.copyTo(mergedBloomFilter)
   tupOlder._2.copyTo(mergedBloomFilter)
   (unionedDf, mergedBloomFilter)
})
println("print combined")
combined._1.show(20, false)

核心修改点 #

• 不再原地修改原有的BloomFilter，而是创建全新的实例，将新旧BF的内容复制进去。
• 每个迭代返回的元组都携带独立的BF实例，确保对应UDF的过滤逻辑不会被后续操作篡改。

修正后，reduceLeft的每次迭代都会生成独立的BF和对应的DataFrame：

1. 第一次迭代：用第三个DF（b: [4,16]）的BF过滤第二个DF，保留c，合并后得到[b, c]的DF和包含b,c的新BF；
2. 第二次迭代：用合并后的DF的BF过滤第一个DF，保留d，最终得到[b,c,d]的DF，符合预期。

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