Spark Bloom Filter 测试

什么是Bloom Filter？

Bloom Filter的原理可参考文章 Bloom Filter原理

由上述文章可知，创建一个Bloom Filter我们只需要定义两个参数：
（1）欲插入Bloom Filter中的元素数目 n
（2）Bloom Filter误判率: P(true)

Bloom Filter的实现

Bloom Filter已经由很多开源库，例如Google guava，Twitter Algebird，ScalaNLP’s Breeze，关于各种Bloom Filter的比较以及改进方法可参考文章 Bloom filter几种实现比较

spark在2.1.0版本也封装了基于DataFrame的Bloom Filter，使用起来相当方便。性能不再受制于第三方框架的吞吐量限制，依赖于spark的并行资源。可以减少架构设计的复杂度，提高可维护性。在流式计算应用中可以将Bloom Filter做成driver级别的全局变量，在batch结束更新。如果考虑容错，可以将Bloom Filter数据定期持久化到磁盘（hdfs/redis)

创建方式

// col为dataframe字段
// nm为放入bloom filter的数据量
// p为bloom filter的误判率
val bf:BloomFilter = df.stat.bloomFilter(col, nm, p)

在用有些场景布隆过滤器还需要合并，官方也提供了相应的API

mergeInPlace(BloomFilter other):BloomFilter

判定数据是否存在,官方一共提供了四个方法：

mightContain(Object obj),mightContainString(String string),mightContainLong(long long),mightContainBinary(byte[] byte)

将Bloom Filter序列化到磁盘或从磁盘加载Bloom Filter

writeTo(OutputStream out)
readFrom(InputStream in)

Bloom Filter的测试

下面将在构建，执行和准确率等方面对spark bloom filter进行测试

1. 完整代码

import org.apache.spark.rdd.RDD
import org.apache.spark.util.sketch._
import org.apache.spark.sql._object BloomFilterTest {def bloom_filter_create(spark:SparkSession):(BloomFilter,DataFrame) = {//构建bloom_filterval time_start = System.currentTimeMillis()val df = spark.sql("select browser_uniq_id from gdm.gdm_m14_glb_wireless_online_log where dt>='2021-05-01' and browser_uniq_id is not null")val nm = df.countprintln(nm)val bf:BloomFilter = df.stat.bloomFilter("browser_uniq_id", nm, 0.01)val time_end = System.currentTimeMillis()println(time_end - time_start)val tuple = (bf,df)tuple}def df_for_test(spark:SparkSession):DataFrame = {//构建测试数据dataframeval df2 = spark.sql("select browser_uniq_id from gdm.gdm_m14_glb_wireless_online_log where dt='2021-04-30'")println(df2.count)df2}def bloom_filter_test(df:DataFrame,bf:BloomFilter): RDD[Row] = {//查询bloom_filterval time_start2=System.currentTimeMillis()val result = df.rdd.map(x=>{if(bf.mightContainString(x.mkString(","))){x}else{Row()}})result.take(1)val time_end2=System.currentTimeMillis()println(time_end2-time_start2)result}def get_error_rate(spark:SparkSession,df1:DataFrame,df2:DataFrame,result:RDD[Row]):Double = {//计算误判率import spark.implicits._val df3 = df1.alias("df1").join(df2.alias("df2"),$"df1.browser_uniq_id"===$"df2.browser_uniq_id","inner")val bf_count = result.filter(x=>x!=Row()).countval df_count = df3.countval error_rate:Double = (bf_count-df_count)*1.0000/df_counterror_rate}def main(args: Array[String]): Unit = {val spark = SparkSession.builder.appName("bbb").master("local[2]").enableHiveSupport().getOrCreate()val bf_tuple = bloom_filter_create(spark)val bf = bf_tuple._1val df1 = bf_tuple._2.distinct()val df2 = df_for_test(spark)val result = bloom_filter_test(df2,bf)println(get_error_rate(spark,df1,df2,result))}
}

2. 测试结果

构建bloom filter数据量：202165350
构建bloom filter耗时：252.356s
测试总数据量：13700749
单条数据判断耗时：190ms
全部数据判断总耗时：6.47s
误判率：1.4222518579818676E-5

3. 结论

在2亿数据量下，构建bloom filter耗时252秒，在初始化bloom filter后可以通过put方法更新，因此在初始化的时候参数nm建议设置的偏大，以便在实际应用过程中追加数据

put(Object item)putString(String item);putLong(long item);putBinary(byte[] item);

通过测试数据可知，spark自带的bloom filter在2亿条基础数据，1千三百万测试数据下，误判率只有1.4222518579818676E-5，个人认为已经非常精准了

实际应用中，在可以接受一定误判的场景下，可以使用bloom filter提高计算效率，例如：

离线数据历史刷数/增量更新
实时数据过滤/排重