数据去重(data deduplication)是大数据领域司空见惯的问题了。除了统计UV等传统用法之外,去重的意义更在于消除不可靠数据源产生的脏数据——即重复上报数据或重复投递数据的影响,使计算产生的结果更加准确。
介绍下经常使用的去重方案:
一、布隆过滤器(BloomFilter)
基本原理:
BloomFilter是由一个长度为m比特的位数组(bit array)与k个哈希函数(hash function)组成的数据结构。位数组均初始化为0,所有哈希函数都可以分别把输入数据尽量均匀地散列。当要插入一个元素时,将其数据分别输入k个哈希函数,产生k个哈希值。以哈希值作为位数组中的下标,将所有k个对应的比特置为1。当要查询(即判断是否存在)一个元素时,同样将其数据输入哈希函数,然后检查对应的k个比特。如果有任意一个比特为0,表明该元素一定不在集合中。如果所有比特均为1,表明该集合有(较大的)可能性在集合中。为什么不是一定在集合中呢?因为一个比特被置为1有可能会受到其他元素的影响,这就是所谓“假阳性”(false positive)。相对地,“假阴性”(false negative)在BloomFilter中是绝不会出现的。
实现参考:
1、Guava中的布隆过滤器:com.google.common.hash.BloomFilter类
2、开源java实现:https://github.com/Baqend/Orestes-Bloomfilter
Redis Bloom Filter扩展:
基于redis做存储后端的BloomFilter实现,可以将bit位存储在redis中,防止计算任务在重启后,当前状态丢失的问题。
二、HyperLogLog(HLL)
HyperLogLog是去重计数的利器,能够以很小的精确度误差作为trade-off大幅减少内存空间占用,在不要求100%准确的计数场景下常用。
HLL基本原理:
- HyperLogLog,以下简称 HLL,它的空间复杂度非常低(log(log(n)) ,故而得名 HLL),几乎不随存储集合的大小而变化;根据精度的不同,一个 HLL 占用的空间从 1KB 到 64KB 不等。而 Bitmap 因为需要为每一个不同的 id 用一个 bit 位表示,所以它存储的集合越大,所占用空间也越大;存储 1 亿内数字的原始 bitmap,空间占用约为 12MB。可以看到,Bitmap 的空间要比 HLL 大约一两个数量级。
- HLL 支持各种数据类型作为输入,使用方便;Bitmap 只支持 int/long 类型的数字作为输入,因此如果原始值是 string 等类型的话,用户需要自己提前进行到 int/long 的映射。
- HLL 之所以支持各种数据类型,是因为其采用了哈希函数,将输入值映射成一个二进制字节,然后对这个二进制字节进行分桶以及再判断其首个1出现的最后位置,来估计目前桶中有多少个不同的值。由于使用了哈希函数,以及使用概率估计的方式,因此 HLL 算法的结果注定是非精确的;尽管 HLL 采用了多种纠正方式来减小误差,但无法改变结果非精确的事实,即便最高精度,理论误差也超过了 1%。
在用Flink做实时计算的过程中,可以用HLL去重计数,比如统计UV。
实现参考:
https://github.com/aggregateknowledge/java-hll
结合Flink,下面的聚合函数即可实现从WindowedStream按天、分key统计PV和UV。
WindowedStream<AnalyticsAccessLogRecord, Tuple, TimeWindow> windowedStream = watermarkedStream .keyBy("siteId") .window(TumblingEventTimeWindows.of(Time.days(1))) .trigger(ContinuousEventTimeTrigger.of(Time.seconds(10))); windowedStream.aggregate(new AggregateFunction<AnalyticsAccessLogRecord, Tuple2<Long, HLL>, Tuple2<Long, Long>>() { private static final long serialVersionUID = 1L; @Override public Tuple2<Long, HLL> createAccumulator() { return new Tuple2<>(0L, new HLL(14, 6)); } @Override public Tuple2<Long, HLL> add(AnalyticsAccessLogRecord record, Tuple2<Long, HLL> acc) { acc.f0++; acc.f1.addRaw(record.getUserId()); return acc; } @Override public Tuple2<Long, Long> getResult(Tuple2<Long, HLL> acc) { return new Tuple2<>(acc.f0, acc.f1.cardinality()); } @Override public Tuple2<Long, HLL> merge(Tuple2<Long, HLL> acc1, Tuple2<Long, HLL> acc2) { acc1.f0 += acc2.f0; acc1.f1.union(acc2.f1); return acc1; } });
三、Roaring Bitmap
布隆过滤器和HyperLogLog,虽然它们节省空间并且效率高,但也付出了一定的代价,即:
- 只能插入元素,不能删除元素;
- 不保证100%准确,总是存在误差。
这两个缺点可以说是所有概率性数据结构(probabilistic data structure)做出的trade-off,毕竟鱼与熊掌不可兼得。
如果一定追求100%准确,普通的位图法显然不合适,应该采用压缩位图(Roaring Bitmap)。
基本原理:
将32位无符号整数按照高16位分桶,即最多可能有216=65536个桶,称为container。存储数据时,按照数据的高16位找到container(找不到就会新建一个),再将低16位放入container中。也就是说,一个RBM就是很多container的集合。依据不同的场景,有 3 种不同的 Container,分别是 Array Container、Bitmap Container 和 Run Container,它们分别通过不同的压缩方法来压缩。实践证明,Roaring Bitmap 可以显著减小 Bitmap 的存储空间和内存占用。
实现参考:
https://github.com/RoaringBitmap/RoaringBitmap
使用限制:
- 对去重的字段只能用整型:int或者long类型,如果要对字符串去重,需要构建一个字符串和整型的映射。
- 对于无法有效压榨的字段(如随机生成的),占用内存较大。
四、外部存储去重
利用外部K-V数据库(Redis、HBase之类)存储需要去重的键。由于外部存储对内存和磁盘占用同样敏感,所以也得设定相应的TTL,以及对大的键进行压缩。另外,外部K-V存储毕竟是独立于应用之外的,一旦计算任务出现问题重启,外部存储的状态和内部状态的一致性(是否需要同步)也是要注意的。
外部存储去重,比如Elasticsearch的 _id 就可以做“去重”功能,但是这种去重的只能针对少量低概率的数据,对全量数据去重是不合适的,因为对ES会产生非常大的压力。
参考:
高效压缩位图RoaringBitmap的原理与应用
谈谈三种海量数据实时去重方案
Flink基于RoaringBitmap的精确去重方案