海量数据处理利器 Roaring BitMap 原理介绍_AI阅读总结 — 包阅AI

在进行大数据开发时，我们可以使用布隆过滤器和Redis中的HyperLogLog来进行大数据的判重和数量统计，虽然这两种方法节省内存空间并且效率很高，但是也存在一些误差。如果需要100%准确的话，我们可以使用BitMap来存储数据。

BitMap 位图索引数据结构被广泛地应用于数据存储和数据搜索中，但是对于存储较为分散的数据时，BitMap会占用比较大的内存空间，因此我们更偏向于使用 Roaring BitMap稀疏位图索引进行存储。同时，Roaring BitMap广泛应用于数据库存储和大数据引擎中，例如Hive，Spark，Doris，Kylin等。

下文将分别介绍 BitMap 和 Roaring BitMap 的原理及其相关应用。

BitMap的基本思想就是用bit位来标记某个元素对应的value，而key就是这个元素。

例如，在下图中，是一个字节代表的8位，下标为1,2,4,6的bit位的值为1，则该字节表示{1,2,4,6}这几个数。

在Java中，1个int占用4个字节，如果用int来存储这四个数字的话，那么将需要4 * 4 = 16字节。

BitMap可以用于快速排序，查找，及去重等操作。优点是占用内存少（相较于数组）和运算效率高，但是缺点也非常明显，无法存储重复的数据，并且在存储较为稀疏的数据时，浪费存储空间较多。

为了解决BitMap存储较为稀疏数据时，浪费存储空间较多的问题，我们引入了稀疏位图索引Roaring BitMap。Roaring BitMap 有较高的计算性能及压缩效率。下面简单介绍一下Roaring BitMap的基本原理。

Roaring BitMap处理int型整数，将32位的int型整数分为高16位和低16位分别进行处理，高16位作为索引分片，而低16位用于存储实际数据。其中每个索引对应一个数据桶（bucket），那么一共可以包含2^16 = 65536个数据块。每个数据桶使用container容器来存储低16位的部分，每个数据桶最多存储2^16 = 65536个数据。

如上图所示，高16位作为索引查找具体的数据块，当前索引值为0，低16位作为value进行存储。

Roaring BitMap在进行数据存储时，会先根据高16位找到对应的索引key（二分查找），低16位作为key对应的value，先通过key检查对应的container容器，如果发现container不存在的话，就先创建一个key和对应的container，否则直接将低16位存储到对应的container中。

Roaring BitMap的精妙之处在于使用不同类型的container，接下来将对其进行介绍。

2.BitMapContainer

BitMapContainer采用BitMap的原理，就是一个没有经过压缩处理的普通BitMap，适合存储比较稠密的数据，在Java中使用Long数组存储低16位数据，每一个bit位表示一个数字。由于每个container需要存储2^16 = 65536个数据，如果通过BitMap进行存储的话，需要使用2^16个bit进行存储，即8kb的数据空间。

可以从下图中看出ArrayContainer和BitMapContainer的内存空间使用关系，当数据量小于4096时，使用ArrayContainer比较合适，当数据量大于等于4096时，使用BitMapContainer更佳。

3.RunContainer

RunContainer采用Run-Length Encoding 行程长度编码进行压缩，适合存储大量连续数据。Java中使用short数组进行存储。连续bit位程度越高的话越节省存储空间，最佳场景下（65536个数据全为1）只需要存储4字节。最差场景为所有数据都不连续，所有存储数据位置为奇数或者偶数，这种场景需要存储128kb。由于采用二分查找算法定位元素，因此时间复杂度为O(logN)。

public void add(final int x) {    final short hb = Util.highbits(x);    final int i = highLowContainer.getIndex(hb);    if (i >= 0) {      highLowContainer.setContainerAtIndex(i,          highLowContainer.getContainerAtIndex(i).add(Util.lowbits(x)));    } else {      final ArrayContainer newac = new ArrayContainer();      highLowContainer.insertNewKeyValueAt(-i - 1, hb, newac.add(Util.lowbits(x)));    }  }

protected int getIndex(short x) {    if ((size == 0) || (keys[size - 1] == x)) {    return size - 1;  }    return this.binarySearch(0, size, x);}

  if (cardinality == 0 || (cardinality > 0          && toIntUnsigned(x) > toIntUnsigned(content[cardinality - 1]))) {    if (cardinality >= DEFAULT_MAX_SIZE) {      return toBitMapContainer().add(x);    }    if (cardinality >= this.content.length) {      increaseCapacity();    }    content[cardinality++] = x;  } else {    int loc = Util.unsignedBinarySearch(content, 0, cardinality, x);    if (loc < 0) {            if (cardinality >= DEFAULT_MAX_SIZE) {        return toBitMapContainer().add(x);      }      if (cardinality >= this.content.length) {        increaseCapacity();      }      System.arraycopy(content, -loc - 1, content, -loc, cardinality + loc + 1);      content[-loc - 1] = x;      ++cardinality;    }  }  return this;}

当基数为0或者值大于container中的最大值，可以直接添加，因为content数组是有序的，最后一个是最大值。

当基数大于等于默认最大值4096时，ArrayContainer将转换为BitMapContainer。如果基数大于content的数组长度的话，需要将content进行扩容。最后进行赋值即可。

public Container add(final short i) {  final int x = Util.toIntUnsigned(i);  final long previous = BitMap[x / 64];  long newval = previous | (1L << x);   BitMap[x / 64] = newval;  if (USE_BRANCHLESS) {    cardinality += (previous ^ newval) >>> x;  } else if (previous != newval) {    ++cardinality;  }  return this;}

public Container add(short k) {     int index = unsignedInterleavedBinarySearch(valueslength, 0, nbrruns, k);  if (index >= 0) {    return this;  }  index = -index - 2;  if (index >= 0) {    int offset = toIntUnsigned(k) - toIntUnsigned(getValue(index));    int le = toIntUnsigned(getLength(index));    if (offset <= le) {      return this;    }    if (offset == le + 1) {            if (index + 1 < nbrruns) {        if (toIntUnsigned(getValue(index + 1)) == toIntUnsigned(k) + 1) {                    setLength(index,              (short) (getValue(index + 1) + getLength(index + 1) - getValue(index)));          recoverRoomAtIndex(index + 1);          return this;        }      }      incrementLength(index);      return this;    }    if (index + 1 < nbrruns) {            if (toIntUnsigned(getValue(index + 1)) == toIntUnsigned(k) + 1) {                setValue(index + 1, k);        setLength(index + 1, (short) (getLength(index + 1) + 1));        return this;      }    }  }  if (index == -1) {        if (0 < nbrruns) {      if (getValue(0) == k + 1) {        incrementLength(0);        decrementValue(0);        return this;      }    }  }  makeRoomAtIndex(index + 1);  setValue(index + 1, k);  setLength(index + 1, (short) 0);  return this;}

public static void count(Integer inputSize) {         RoaringBitMap BitMap = new RoaringBitMap();         BitMap.add(0L, inputSize);                 int cardinality = BitMap.getCardinality();                 int compressSizeIntBytes = BitMap.getSizeInBytes();                          int uncompressSizeIntBytes = BitMap.getSizeInBytes();         System.out.println("Roaring BitMap个数：" + cardinality);        System.out.println("最好情况，BitMap压缩大小：" + compressSizeIntBytes / 1024 + "KB");        System.out.println("最坏情况，BitMap不压缩大小：" + uncompressSizeIntBytes / 1024 / 1024 + "MB");         BitSet bitSet = new BitSet();        for (int i = 0; i < inputSize; i++) {            bitSet.set(i);        }                int size = bitSet.size();         System.out.println("BitMap个数：" + bitSet.length());        System.out.println("BitMap大小：" + size / 8 / 1024 / 1024 + "MB");    }

public void add(final int x)
public void add(final long rangeStart, final long rangeEnd)
public void remove(final int x)
public void forEach(IntConsumer ic)
public boolean contains(final int x)
public int getCardinality()
public void and(final RoaringBitMap x2)
public static RoaringBitMap and(final RoaringBitMap x1, final RoaringBitMap x2)
public void or(final RoaringBitMap x2)
public static RoaringBitMap or(final RoaringBitMap x1, final RoaringBitMap x2)
public void xor(final RoaringBitMap x2)
public static RoaringBitMap xor(final RoaringBitMap x1, final RoaringBitMap x2)
public void andNot(final RoaringBitMap x2)
public static RoaringBitMap andNot(final RoaringBitMap x1, final RoaringBitMap x2)
public void serialize(DataOutput out) throws IOExceptionpublic void serialize(ByteBuffer buffer)
public void deserialize(DataInput in) throws IOExceptionpublic void deserialize(ByteBuffer bbf) throws IOException

Roaring BitMap可以用来构建大数据标签，针对类型特征来创建对应的标签。

在我们的业务场景中，有很多需要基于人群标签进行交并补集运算的场景，下面以一个场景为例，我们需要计算每天某个设备接口 在设备标签A上的查询成功率，因为设备标签A中的设备不是所有都活跃在网的，所以我们需要将设备标签A与每日日活人群标签取交集，得到的交集大小才能用作成功率计算的分母，另外拿查询成功的标签人群做分子来进行计算即可，查询时长耗时为1s。

假如没有使用标签保存集合之前，我们需要在hive表中查询出同时满足当天在网的活跃用户和设备A的用户数量，查询时长耗时在几分钟以上。两种方式相比之下，使用Roaring BitMap查询的效率更高。