Spark布隆过滤器超4亿元素场景误判率远超预期的原因

问题根源 #

你遇到的高误判率是由于Spark 3.5.0中BloomFilterImpl的整数溢出问题导致的：

• 对于5亿元素、1%误判率的布隆过滤器，理论所需的位数组大小约为48亿位（计算方式：m = -n * ln(fpp) / (ln2)^2，代入n=5e8、fpp=0.01可得）。
• 但Java的int类型最大值仅为21亿位（2^31-1），Spark的BloomFilterImpl使用int存储位数组大小，计算时直接发生整数溢出，导致实际分配的位数远小于理论需求。
• 位数组容量严重不足直接导致误判率急剧上升，这也是小容量过滤器（1亿元素，理论所需位数约9600万位，远小于int上限）表现正常的原因。

验证方法 #

你可以打印布隆过滤器的实际参数，确认溢出问题：

println(s"实际位数: ${bloom_filter.numBits()}")
println(s"实际哈希函数数量: ${bloom_filter.numHashFunctions()}")

理论上，1%误判率下哈希函数数量应为约7个，但溢出后实际值会远偏离这个数值，位数组大小也会远小于48亿位。

解决方案 #

针对这个问题，你可以选择以下几种解决方式：

1. 降低预期误判率阈值：
   调整fpp参数，使理论所需位数不超过int上限。例如，当fpp设为0.2时，5亿元素所需位数约16.8亿位，在int范围内，误判率会接近预期。
2. 使用支持64位位数组的布隆过滤器实现：
   替换为Guava的BloomFilter（支持long类型存储位数），或自定义基于分块的布隆过滤器，避免容量限制。
3. 拆分数据构建多个过滤器：
   将5亿数据拆分为多个子集（如5个1亿元素的子集），分别构建布隆过滤器，查询时检查所有过滤器，只要有一个返回true则判定为存在。
4. 升级Spark版本：
   检查Spark官方修复记录，升级到修复了整数溢出问题的版本。

复现代码 #

____              __
     / __/__  ___ _____/ /__
    _\ \/_ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.5.0-amzn-0
      /_/
         
Using Scala version 2.12.17 (OpenJDK 64-Bit Server VM, Java 17.0.10)
Type in expressions to have them evaluated.
Type :help for more information.

scala> import java.security.MessageDigest
import java.security.MessageDigest

scala> import scala.util.Random
import scala.util.Random

scala> import org.apache.spark.util.sketch.BloomFilter
import org.apache.spark.util.sketch.BloomFilter

scala> // Function to generate a random SHA1 hash

scala> def generateRandomSha1(): String = {
     |   val randomString = Random.alphanumeric.take(20).mkString
     |   val sha1 = MessageDigest.getInstance("SHA-1")
     |   sha1.update(randomString.getBytes("UTF-8"))
     |   val digest = sha1.digest
     |   digest.map("%02x".format(_)).mkString
     | }
generateRandomSha1: ()String

scala> // Generate a DataFrame with 500 million rows of random SHA1 hashes

scala> val df = spark.range(500000000).map(_ => generateRandomSha1()).toDF("Hash")
df: org.apache.spark.sql.DataFrame = [Hash: string]

scala> // Create a bloom filter out of this collection of strings.

scala> val bloom_filter = df.stat.bloomFilter("Hash", 500000000, 0.01)
bloom_filter: org.apache.spark.util.sketch.BloomFilter = org.apache.spark.util.sketch.BloomFilterImpl@a14c0ba9

scala> // Generate another 10,000 random hashes

scala> val random_sha1s = List.fill(10000)(generateRandomSha1())
random_sha1s: List[String] = List(f3cbfd9bd836ea917ebc0dfc5330135cfde322a3, 4bff8d58799e517a1ba78236db9b52353dd39b56, 775bdd9d138a79eeae7308617f5c0d1d0e1c1697, abbd761b7768f3cbadbffc0c7947185856c4943d, 343692fe61c552f73ad6bc2d2d3072cc456da1db, faf4430055c528c9a00a46e9fae7dc25047ffaf3, 255b5d56c39bfba861647fff67704e6bc758d683, dae8e0910a368f034958ae232aa5f5285486a8ac, 3680dbd34437ca661592a7e4d39782c9c77fb4ba, f5b43f7a77c9d9ea28101a1848d8b1a1c0a65b82, 5bda825102026bc0da731dc84d56a499ccff0fe1, 158d7b3ce949422de421d5e110e3f6903af4f8e1, 2efcae5cb10273a0f5e89ae34fa3156238ab0555, 8d241012d42097f80f30e8ead227d75ab77086d2, 307495c98ae5f25026b91e60cf51d4f9f1ad7f4b, 8fc2f55563ab67d4ec87ff7b04a4a01e821814a3, b413572d14ee16c6c575ca3472adff62a8cbfa3d, 9219233b0e8afe57d7d5cb6...

scala> // Check how many of these random hashes return a positive result when passed into mightContain

scala> random_sha1s.map(c => bloom_filter.mightContain(c)).count(_ == true)
res0: Int = 2153

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