BloomFilter mergeInPlace() 导致Spark DataFrame出现意外行为

核心原因 #

Spark的DataFrame采用惰性计算机制，而你在UDF中直接引用了可变的BloomFilter实例a._2。调用mergeInPlace()修改该实例的内部状态后，后续触发DataFrame计算时，UDF会使用修改后的BloomFilter，导致过滤逻辑的输出结果改变，最终让c的两次show()结果不一致。

详细拆解 #

• DataFrame的惰性计算特性：c = a._1.union(filtered_b_df)只是定义了计算逻辑，并没有立即执行数据处理。只有当调用show()这类触发动作时，Spark才会启动Job执行完整的计算链路。
• UDF对可变实例的引用：a_bloom_filter_udf直接绑定了a._2这个BloomFilter实例，而非它的副本。第一次执行c.show()时：
  • a._2仅包含元素"b"，所以b_df中的"b"会被过滤，"c"会被保留，最终c包含a_df的("b", List("4", "16"))和b_df的("c", List("4"))。
• mergeInPlace()的副作用：a._2.mergeInPlace(b._2)会直接修改a._2的内部位图，将b_bloom_filter中的元素（"b"和"c"）合并进去。此时a._2已经包含了"c"。
• 第二次show()的重新计算：第二次调用c.show()时，Spark会重新执行整个计算链路，UDF此时调用!a._2.mightContain(s)：
  • b_df中的"c"现在会被a._2识别为存在，返回false被过滤，filtered_b_df变为空DataFrame，最终c仅保留a_df的数据。

验证逻辑的合理性 #

当你调用b._2.mergeInPlace(a._2)时，修改的是b._2而非a._2，a._2的状态始终保持初始状态（仅包含"b"），所以两次c.show()的过滤逻辑结果一致，符合预期。

建议 #

避免在UDF中引用可变对象，因为Spark的惰性计算会导致后续Job执行时使用修改后的对象状态，引发非预期结果。如果需要使用合并后的BloomFilter，建议创建新的实例，或者在UDF中使用不可变的副本。

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