Spark 和 BloomFilter 共享

Question

我有一个巨大的 RDD（源），我需要从中创建一个 BloomFilter 数据，因此对用户数据的后续更新将只考虑真正的“差异”，没有重复。

看起来 BloomFilter 的大多数实现都是不可序列化的（虽然可以很容易地修复），但我想要稍微不同的工作流程：

1. 处理每个分区并为每个分区创建一个适当的 BloomFilter 实例。对于每个 BloomFilter 对象 - 将其写入某个二进制文件。我实际上不知道如何处理整个分区 - RDD 上有 mapPartition 函数可用，但这希望我返回一个迭代器。也许我可以使用传递的迭代器，创建一个 BloomFilter 的实例，将其写入某个位置并将指向创建文件的链接作为 Iterator.singleton[PathToFile] 返回？
2. 在主节​​点 - consume 该处理的结果（文件的路径列表），读取这些文件并在内存中聚合 BloomFilter。然后将响应写入二进制文件。

我不知道正确的方法：

• 在传递给mapPartitions的函数中，在集群支持的FS中创建一个文件（可以是HDFS、S3N或本地文件）
• 在第二阶段使用consume 读取文件的内容（当我有一个带有文件路径的RDD 时，我必须使用SparkContext 来读取它们 - 不知道怎么可能） .

谢谢！

Answer 1

breeze 实现不是最快的，但它带有通常的 Spark 依赖项，可以与 simple aggregate 一起使用：

import breeze.util.BloomFilter

// Adjust values to fit your case
val numBuckets: Int = 100
val numHashFunctions: Int = 30

val rdd = sc.parallelize(Seq("a", "d", "f", "e", "g", "j", "z", "k"), 4)
val bf = rdd.aggregate(new BloomFilter[String](numBuckets, numHashFunctions))(
  _ += _, _ |= _
)

bf.contains("a")

Boolean = true

bf.contains("n")

Boolean = false

在 Spark 2.0+ 中你可以使用DataFrameStatFunctions.bloomFilter:

val df = rdd.toDF

val expectedNumItems: Long = 1000 
val fpp: Double = 0.005

val sbf = df.stat.bloomFilter($"value", expectedNumItems, fpp)

sbf.mightContain("a")

Boolean = true

sbf.mightContain("n")

Boolean = false

Algebird 实现也可以工作，并且可以与breeze 实现类似地使用。