math-doc-布隆过滤

介绍

有时候我们需要判断一个元素是否在一个集合中。比如，在字处理软件中，需要检查一个单词是否拼写正确（也就是要判断它是否在已知的字典里）；在警察系统中，一个嫌疑人的名字是否出现在嫌疑名单上；在网络爬虫里，一个网址是否已经被访问过，等等。

最直接的方法就是讲集合中的元素存在计算机中，遇到一个新元素时，将它和集合中的元素直接比较即可。一般来讲，计算机中的集合是用哈希表（Hash Table）来存储的。它的好处是快速准确，缺点是耗费存储空间。

为什么说耗费存储空间呢？其根本原因是哈希表方法需要把实实在在的具有特定长度（每个Email地址对应成一个8字节的信息指纹）的元素的信息指纹存储在内存或硬盘中的哈希表中，这个存储量在实际应用中一般是相当大的。比如每存储一亿个Email地址，需要0.8G大小的数字指纹存储空间，考虑到哈希表的存储空间利用率一般只有一半，所以需要1.6G的存储空间。如果存储几十亿上百亿的Email地址，那就需要百亿字节的内存存储空间。

而布隆过滤器只需要哈希表1/8到1/4的大小就能解决同样的问题，它实际上是一个很长的二进制向量和一系列的随机映射函数。

算法描述

一个empty bloom filter是一个有m bits的bit array，每一个bit位都初始化为0。并且定义有k个不同的hash function，每个都以uniform random distribution(均匀随机分布)将元素hash到m个不同位置中的一个。

在下面的介绍中n为元素数，m为布隆过滤器或哈希表的slot数，k为布隆过滤器重hash function数。

为了add一个元素，用k个hash function将它hash得到bloom filter中k个bit位，将这k个bit位置1。

为了query一个元素，即判断它是否在集合中，用k个hash function将它hash得到k个bit位。若这k bits全为1，则此元素在集合中；若其中任一位不为1，则此元素比不在集合中（因为如果在，则在add时已经把对应的k个bits位置为1）。

不允许remove元素，因为那样的话会把相应的k个bits位置为0，而其中很有可能有其他元素对应的位。因此remove会引入false negative，这是绝对不被允许的。

当k很大时，设计k个独立的hash function是不现实并且困难的。对于一个输出范围很大的hash function（例如MD5产生的128 bits数），如果不同bit位的相关性很小，则可把此输出分割为k份。或者可将k个不同的初始值（例如0,1,2, … ,k-1）结合元素，feed给一个hash function从而产生k个不同的数。

当add的元素过多时，即n/m过大时（n是元素数，m是bloom filter的bits数），会导致false positive过高(重复过高)，此时就需要重新组建filter，但这种情况相对少见，通常都是一开始就估算一个比实际要大的相关参数。

时间和空间上的优势

当可以承受一些误报时，布隆过滤器比其它表示集合的数据结构有着很大的空间优势。例如self-balance BST, tries, hash table或者array, chain，它们中大多数至少都要存储元素本身，对于小整数需要少量的bits，对于字符串则需要任意多的bits（tries是个例外，因为对于有相同prefixes的元素可以共享存储空间）；而chain结构还需要为存储指针付出额外的代价。对于一个有1%误报率和一个最优k值的布隆过滤器来说，无论元素的类型及大小，每个元素只需要9.6 bits来存储。这个优点一部分继承自array的紧凑性，一部分来源于它的概率性。如果你认为1%的误报率太高，那么对每个元素每增加4.8 bits，我们就可将误报率降低为原来的1/10。add和query的时间复杂度都为O(k)，与集合中元素的多少无关，这是其他数据结构都不能完成的。

如果可能元素范围不是很大，并且大多数都在集合中，则使用确定性的bit array远远胜过使用布隆过滤器。因为bit array对于每个可能的元素空间上只需要1 bit，add和query的时间复杂度只有O(1)。注意到这样一个哈希表（bit array）只有在忽略collision并且只存储元素是否在其中的二进制信息时，才会获得空间和时间上的优势，而在此情况下，它就有效地称为了k=1的布隆过滤器。

而当考虑到collision时，对于有m个slot的bit array或者其他哈希表（即k=1的布隆过滤器），如果想要保证1%的误判率，则这个bit array只能存储m/100个元素，因而有大量的空间被浪费，同时也会使得空间复杂度急剧上升，这显然不是space efficient的。解决的方法很简单，使用k>1的布隆过滤器，即k个hash function将每个元素改为对应于k个bits，因为误判度会降低很多，并且如果参数k和m选取得好，一半的m可被置为为1，这充分说明了布隆过滤器的space efficient性。

举例说明

以垃圾邮件过滤中黑白名单为例：现有1亿个email的黑名单，每个都拥有8 bytes的指纹信息，则可能的元素范围为 $2^{64}=8*10^{18}bits=10^9GB$，对于bit array来说是根本不可能的范围，而且元素的数量（即email列表）为 $10^8$ ，相比于元素范围过于稀疏，而且还没有考虑到哈希表中的collision问题。

若采用哈希表，由于大多数采用open addressing来解决collision，而此时的search时间复杂度为 ：
$$\dfrac{1}{1-{\dfrac{n}{m}}}$$

即若哈希表半满$(\dfrac{n}{m}=\dfrac{1}{2})$，则每次search需要probe 2次，因此在保证效率的情况下哈希表的存储效率最好不超过50%。此时每个元素占8 bytes，总空间为：
$$\dfrac{10^8*8bytes}{50%}=1.6GB$$

若采用Perfect hashing（这里可以采用Perfect hashing是因为主要操作是search/query，而并不是add和remove），虽然保证worst-case也只有一次probe，但是空间利用率更低，一般情况下为50%，worst-case时有不到一半的概率为25%。

若采用布隆过滤器，取k=8。因为n为1亿，所以总共需要 $810^8bits$ 被置位为1，又因为在保证误判率低且k和m选取合适时，空间利用率为50%，所以总空间为：
$$m=\dfrac{10^88bits}{50%}=1.6*10^9bits=200MB$$

所需空间比上述哈希结构小得多，并且误判率在万分之一以下。

误判概率的证明和计算

假设布隆过滤器中的hash function满足simple uniform hashing假设：每个元素都等概率地hash到m个slot中的任何一个，与其它元素被hash到哪个slot无关。若m为bit数，则对某一特定bit位在一个元素由某特定hash function插入时没有被置位为1的概率为：$1-\dfrac{1}{m}$，则k个hash function中没有一个对其置位的概率为：$(1-\dfrac{1}{m})^k$

如果插入了n个元素，但都未将其置位的概率为：$(1-\dfrac{1}{m})^{kn}$，则此位被置位的概率为：$1-(1-\dfrac{1}{m})^{kn}$

现在考虑query阶段，若对应某个待query元素的k bits全部置位为1，则可判定其在集合中。因此将某元素误判的概率为：$(1-(1-\dfrac{1}{m})^{kn})^k$

由于$(1+x)^{\dfrac{1}{x}}≈e$，当$x\implies 0$时，并且$-\dfrac{1}{m}$在当m很大时趋近于0，所以
$$(1-(1-\dfrac{1}{m})^{kn})^k=(1-(1-\dfrac{1}{m})^{-m\dfrac{-kn}{m}})^k≈(1-e^{\dfrac{-nk}{m}})^k$$

从上式中可以看出，当m增大或n减小时，都会使得误判率减小。

现在计算对于给定的m和n，k为何值时可以使得误判率最低。设误判率为k的函数为：$f(x)=(1-e^{\dfrac{-nk}{m}})^k$。设$b=e{\dfrac{n}{m}}$，则简化为$f(x)=(1-b^{-k})^k$，两边取对数得：$\ln{f(x)}=k*\ln{(1-b^{-k})}$，两边对k求导得：

$$
\dfrac{1}{f(x)}f’(x)=\ln{(1-b^{-k})}+k*\dfrac{1}{1-b^{-k}}(-b^{-k})\ln{b}(-1)\
=\ln{(1-b^{-k})}+k\dfrac{b^{-k}\ln{b}}{1-b^{-k}}
$$

下面求最值
$$
\ln{(1-b^{-k})+k\dfrac{b{-k}\ln{b}}{1-b^{-k}}}=0\
\implies (1-b^{-k})\ln{(1-b^{-k})}=-kb^{-k}\ln{b}\
\implies (1-b^{-k})\ln{(1-b^{-k})}=b^{-k}\ln{b^{-k}}\
\implies 1-b^{-k}=b^{-k}\
\implies b^{-k}=\dfrac{1}{2}\
\implies e^{-\dfrac{kn}{m}}=\dfrac{1}{2}\
\implies \dfrac{kn}{m}=\ln{2}\
\implies k=\ln{2}\dfrac{m}{n}=0.7*\dfrac{m}{n}
$$

因此，当$k=0.7*\dfrac{m}{n}$时误判率最低，次数误判率为：$p(error)=(1-\dfrac{1}{2})^k=2^{-k}=2^{-\ln{2}\dfrac{m}{n}}≈0.6185^{\dfrac{m}{n}}$

可以看出若要使得误判率$≤\dfrac{1}{2}$，则：$k≥1\implies \dfrac{m}{n}≥\dfrac{1}{\ln{2}}$

这说明了若想保持某固定误判率不变，布隆过滤器的bit数m与被add的元素数n应该是线性同步增加的。

设计和应用布隆过滤器的方法

应用时首先要先由用户决定要add的元素数n和希望的误差率P。这也是一个设计完整的布隆过滤器需要用户输入的仅有的两个参数，之后的所有参数将由系统计算，并由此建立布隆过滤器。

系统首先要计算需要的内存大小m bits:
$$p=2^{\ln{2}\dfrac{m}{n}}\implies \ln{p}=\ln{2}(-\ln{2})\dfrac{m}{n}\implies m=-\dfrac{n\ln{p}}{(\ln{2})^2}$$

再由m，n得到hash function的个数：
$$k=\ln{2}\dfrac{m}{n}=0.7\dfrac{m}{n}$$

至此系统所需的参数已经备齐，接下来add n个元素至布隆过滤器中，再进行query。

根据公式，当k最优时：
$$
p(error)=2^{-k}\
\implies\log{_2^{\dfrac{1}{p}}}=-k\
\implies k=\log{_2^\dfrac{1}{p}}\
\implies \ln{2}\dfrac{m}{n}=\log{_2^\dfrac{1}{p}}\
\implies\dfrac{m}{n}=\ln{2}\log{_2^\dfrac{1}{p}}=1.44*\log{_2^\dfrac{1}{p}}
$$

因此可验证当P=1%时，存储每个元素需要9.6 bits：$\dfrac{m}{n}=1.44*\log{_2^{\dfrac{1}{0.01}}}=9.6bits$

而每当想将误判率降低为原来的1/10，则存储每个元素需要增加4.8 bits：
$$\dfrac{m}{n}=1.44(\log{_2^{10a}}-\log{_2^a})=1.44\log{_2^10}=4.8bits$$

这里需要特别注意的是，9.6 bits/element不仅包含了被置为1的k位，还把包含了没有被置为1的一些位数。此时的 $$k=0.7\dfrac{m}{n}=0.79.6=6.72bits$$ 才是每个元素对应的为1的bit位数。

$k=0.7*\dfrac{m}{n}$从而使得p(error)最小时，我们注意到：$p(error)=(1-e^{-\dfrac{nk}{m}})^k$中的$e^{-\dfrac{nk}{m}}=\dfrac{1}{2}$，即$(1-\dfrac{1}{m})^{kn}=\dfrac{1}{2}$

此概率为某bit位在插入n个元素后未被置位的概率。因此，想保持错误率低，布隆过滤器的空间使用率需为50%。

简单实现

import java.io.Serializable;
import java.nio.charset.Charset;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.BitSet;
import java.util.Collection;

/**
 * 布隆过滤器的实现类
 * 定义：http://en.wikipedia.org/wiki/Bloom_filter
 * 
 * @param <E>
 *            E指定了要插入过滤器中的元素的类型，eg，String Integer
 * @author Magnus Skjegstad <magnus@skjegstad.com>
 * @translator xiaoye
 */
public class BloomFilter<E> implements Serializable {

    private static final long serialVersionUID = -2326638072608273135L;
    private BitSet bitset;
    private int bitSetSize;
    private double bitsPerElement;
    private int expectedNumberOfFilterElements;//能够添加的元素的最大个数 
    private int numberOfAddedElements;//过滤器容器中元素的实际数量 
    private int k; // 哈希函数的个数

    static final Charset charset = Charset.forName("UTF-8");//存储哈希值的字符串的编码方式

    static final String hashName = "MD5"; //在大多数情况下，MD5提供了较好的散列精确度。如有必要，可以换成 SHA1算法
    static final MessageDigest digestFunction;//MessageDigest类用于为应用程序提供信息摘要算法的功能，如 MD5 或 SHA 算法
    static { // 初始化 MessageDigest 的摘要算法对象
        MessageDigest tmp;
        try {
            tmp = java.security.MessageDigest.getInstance(hashName);
        } catch (NoSuchAlgorithmException e) {
            tmp = null;
        }
        digestFunction = tmp;
    }

    /**
     * 构造一个空的布隆过滤器. 过滤器的长度为c*n
     * @param c
     *            表示每个元素占有多少位
     * @param n
     *            表示过滤器能添加的最大元素数量
     * @param k
     *            表示需要使用的哈希函数的个数
     */
    public BloomFilter(double c, int n, int k) {
        this.expectedNumberOfFilterElements = n;
        this.k = k;
        this.bitsPerElement = c;
        this.bitSetSize = (int) Math.ceil(c * n);
        numberOfAddedElements = 0;
        this.bitset = new BitSet(bitSetSize);
    }

    /**
     * 构造一个空的布隆过滤器。最优哈希函数的个数将由过滤器的总大小和期望元素个数来确定。
     * 
     * @param bitSetSize
     *            指定了过滤器的总大小
     * @param expectedNumberOElements
     *            指定了过滤器能添加的最大的元素数量
     */
    public BloomFilter(int bitSetSize, int expectedNumberOElements) {
        this(bitSetSize / (double) expectedNumberOElements,  expectedNumberOElements, (int) Math.round((bitSetSize / (double) expectedNumberOElements)* Math.log(2.0)));
    }

    /**
     * 通过指定误报率来构造一个过滤器。
     * 每个元素所占的位数和哈希函数的数量会根据误报率来得出。
     * 
     * @param falsePositiveProbability
     *            所期望误报率.
     * @param expectedNumberOfElements
     *            要添加的元素的数量
     */
    public BloomFilter(double falsePositiveProbability, int expectedNumberOfElements) {
        this(Math.ceil(-(Math.log(falsePositiveProbability) / Math.log(2)))/ Math.log(2), // c = k/ln(2)
                expectedNumberOfElements, 
                (int) Math.ceil(-(Math.log(falsePositiveProbability) / Math.log(2)))); // k = ln(2)m/n
    }

    /**
     * 根据旧过滤器的数据，重新构造一个新的过滤器
     * 
     * @param bitSetSize
     *            指定了过滤器所需位的大小
     * @param expectedNumberOfFilterElements
     *            指定了过滤器所能添加的元素的最大数量
     *            to contain.
     * @param actualNumberOfFilterElements
     *            指定了原来过滤器的数据的数量
     *            <code>filterData</code> BitSet.
     * @param filterData
     *            原有过滤器中的BitSet对象
     */
    public BloomFilter(int bitSetSize, int expectedNumberOfFilterElements,
            int actualNumberOfFilterElements, BitSet filterData) {
        this(bitSetSize, expectedNumberOfFilterElements);
        this.bitset = filterData;
        this.numberOfAddedElements = actualNumberOfFilterElements;
    }

    /**
     * 根据字符串的内容生成摘要
     * 
     * @param val
     *            字符串的内容
     * @param charset
     *            输入数据的编码方式
     * @return    输出为一个long类型
     */
    public static long createHash(String val, Charset charset) {
        return createHash(val.getBytes(charset));
    }

    /**
     * 根据字符串内容生成摘要
     * 
     * @param val
     *            指定了输入的字符串。默认的编码为 UTF-8
     * @return 输出为一个long类型
     */
    public static long createHash(String val) {
        return createHash(val, charset);
    }

    /**
     * 根据字节数组生成摘要
     * 
     * @param data
     *            输入数据
     * @return 输出为long类型的摘要
     */
    public static long createHash(byte[] data) {
        long h = 0;
        byte[] res;

        synchronized (digestFunction) {
            res = digestFunction.digest(data);
        }

        for (int i = 0; i < 4; i++) {
            h <<= 8;
            h |= ((int) res[i]) & 0xFF;
        }
        return h;
    }

    /**
     * 重写equals方法
     */
    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final BloomFilter<E> other = (BloomFilter<E>) obj;
        if (this.expectedNumberOfFilterElements != other.expectedNumberOfFilterElements) {
            return false;
        }
        if (this.k != other.k) {
            return false;
        }
        if (this.bitSetSize != other.bitSetSize) {
            return false;
        }
        if (this.bitset != other.bitset
                && (this.bitset == null || !this.bitset.equals(other.bitset))) {
            return false;
        }
        return true;
    }

    /**
     * 重写了hashCode方法
     * 
     */
    @Override
    public int hashCode() {
        int hash = 7;
        hash = 61 * hash + (this.bitset != null ? this.bitset.hashCode() : 0);
        hash = 61 * hash + this.expectedNumberOfFilterElements;
        hash = 61 * hash + this.bitSetSize;
        hash = 61 * hash + this.k;
        return hash;
    }

    /**
     * 根据最大元素数量和过滤器的大小来计算误报率。
     * 方法的返回值为误报率。如果插入的元素个数小于最大值，则误报率会比返回值要小。
     * 
     * @return 期望的误报率.
     */
    public double expectedFalsePositiveProbability() {
        return getFalsePositiveProbability(expectedNumberOfFilterElements);
    }

    /**
     * 通过插入的元素数量和过滤器容器大小来计算实际的误报率。
     * 
     * @param numberOfElements
     *            插入的元素的个数.
     * @return 误报率.
     */
    public double getFalsePositiveProbability(double numberOfElements) {
        // (1 - e^(-k * n / m)) ^ k
        return Math.pow((1 - Math.exp(-k * (double) numberOfElements
                / (double) bitSetSize)), k);

    }

    /**
     * 通过实际插入的元素数量和过滤器容器大小来计算实际的误报率。
     * 
     * @return 误报率.
     */
    public double getFalsePositiveProbability() {
        return getFalsePositiveProbability(numberOfAddedElements);
    }

    /**
     * 返回哈希函数的个数 k 
     * 
     * @return  k.
     */
    public int getK() {
        return k;
    }

    /**
     * 清空过滤器元素
     */
    public void clear() {
        bitset.clear();
        numberOfAddedElements = 0;
    }

    /**
     * 向过滤器中添加元素。
     * 添加的元素的toString()方法将会被调用，返回的字符串作为哈希函数的输出。
     * 
     * @param element
     *            要添加的元素
     */
    public void add(E element) {
        long hash;
        String valString = element.toString();
        for (int x = 0; x < k; x++) {
            hash = createHash(valString + Integer.toString(x));
            hash = hash % (long) bitSetSize;
            bitset.set(Math.abs((int) hash), true);
        }
        numberOfAddedElements++;
    }

    /**
     * 添加一个元素集合到过滤器中
     * 
     * @param c
     *            元素集合.
     */
    public void addAll(Collection<? extends E> c) {
        for (E element : c)
            add(element);
    }

    /**
     * 用来判断元素是否在过滤器中。如果已存在，返回 true。
     * 
     * @param element
     *            要检查的元素.
     * @return 如果估计该元素已存在，则返回true
     */
    public boolean contains(E element) {
        long hash;
        String valString = element.toString();
        for (int x = 0; x < k; x++) {
            hash = createHash(valString + Integer.toString(x));
            hash = hash % (long) bitSetSize;
            if (!bitset.get(Math.abs((int) hash)))
                return false;
        }
        return true;
    }

    /**
     * 判断一个集合中的元素是否都在过滤器中。
     * 
     * @param c
     *            要检查的元素集合
     * @return 如果集合所有的元素都在过滤器中，则返回true。
     */
    public boolean containsAll(Collection<? extends E> c) {
        for (E element : c)
            if (!contains(element))
                return false;
        return true;
    }

    /**
     * 得到某一位的值
     * 
     * @param bit
     *            bit的位置.
     * @return 如果该位被设置，则返回true。
     */
    public boolean getBit(int bit) {
        return bitset.get(bit);
    }

    /**
     * 设置过滤器某一位的值
     * 
     * @param bit
     *            要设置的位置.
     * @param value
     *            true表示已经成功设置。false表示改为被清除。
     */
    public void setBit(int bit, boolean value) {
        bitset.set(bit, value);
    }

    /**
     * 返回存放信息的位数组.
     * 
     * @return 位数组.
     */
    public BitSet getBitSet() {
        return bitset;
    }

    /**
     * 得到过滤器中位数组个大小。
     * 
     * @return 数组大小.
     */
    public int size() {
        return this.bitSetSize;
    }

    /**
     * 返回已添加的元素的个数
     * 
     * @return 元素个数.
     */
    public int count() {
        return this.numberOfAddedElements;
    }

    /**
     * 得到能添加的元素的最大数量
     * 
     * @return  最大数量.
     */
    public int getExpectedNumberOfElements() {
        return expectedNumberOfFilterElements;
    }

    /**
     * 得到每个元素占用的位的个数的期望值
     * 
     * @return 每个元素占用的位数
     */
    public double getExpectedBitsPerElement() {
        return this.bitsPerElement;
    }

    /**
     * 得到每个元素占用位数的实际值
     * 
     * @return 每个元素占用的位数.
     */
    public double getBitsPerElement() {
        return this.bitSetSize / (double) numberOfAddedElements;
    }
}