Réduction avec un filtre de bloom

Question

Je souhaite obtenir une appartenance à un ensemble approximatif rapide, basée sur une fonction de chaîne de caractères appliquée à un grand RDD Spark de vecteurs de chaîne (enregistrements ~ 1B). Fondamentalement, l'idée serait de réduire en Bloom filter. Ce filtre bloom pourrait ensuite être diffusé aux travailleurs pour une utilisation ultérieure.

Plus précisément, j'ai actuellement

rdd: RDD[Vector[String]] 
f: Vector[String] => String 
val uniqueVals = rdd.map(f).distinct().collect() 
val uv = sc.broadcast(uniqueVals) 

Mais uniqueVals est trop grand pour être pratique, et je voudrais le remplacer par quelque chose de plus petite taille (et connu), à savoir un filtre bloom.

Mes questions:

• est-il possible de réduire dans un filtre Bloom, ou dois-je recueillir d'abord, puis le construire dans le pilote?

• Y a-t-il une implémentation de filtre Scala/Java Bloom mature qui conviendrait?

Answer 1

Oui, les filtres Bloom peuvent être réduits, car ils ont quelques bonnes propriétés (ils sont monoids). Cela signifie que vous pouvez effectuer toutes les opérations d'agrégation en parallèle, en effectuant un seul passage sur les données pour construire le BloomFilter pour chaque partition, puis réduire ces BloomFilters ensemble pour obtenir un seul BloomFilter pour lequel vous pouvez demander contains.

Il existe au moins deux implémentations de BloomFilter dans Scala et les deux semblent des projets matures (ne les ont pas réellement utilisés en production). Le premier est Breeze et le second est Twitter's Algebird. Les deux contiennent des implémentations de croquis différents et beaucoup plus.

Voici un exemple comment faire avec Breeze:

import breeze.util.BloomFilter 

val nums = List(1 to 20: _*).map(_.toString) 
val rdd = sc.parallelize(nums, 5) 

val bf = rdd.mapPartitions { iter => 
    val bf = BloomFilter.optimallySized[String](10000, 0.001) 
    iter.foreach(i => bf += i) 
    Iterator(bf) 
}.reduce(_ | _) 

println(bf.contains("5")) // true 
println(bf.contains("31")) // false