在工作中经常会遇到去重的场景，例如基于 App 的用户行为日志分析系统，用户的行为日志从手机客户端上报到 Nginx 服务端，通过 Logstash、Flume 或其他工具将日志从 Nginx 写入到 Kafka 中。

由于用户手机客户端的网络可能出现不稳定，所以手机客户端上传日志的策略是：宁可重复上报，也不能丢日志。所以导致 Kafka 中必然会出现日志重复的情况，即：同一条日志出现了 2 条或 2 条以上。

通常情况下，Flink 任务的数据源都是 Kafka，若 Kafka 中数据出现了重复，在实时 ETL 或者流计算时都需要考虑对日志主键进行去重，否则会导致流计算结果偏高或结果不准确的问题，例如用户 a 在某个页面只点击了一次，但由于日志重复上报，所以用户 a 在该页面的点击日志在 Kafka 中出现了 2 次，最后统计该页面的 click 数时，结果就会偏高。

这里只阐述了一种可能造成 Kafka 中数据重复的情况，在生产环境中很多情况都可能造成 Kafka 中数据重复，这里不一一列举，本节主要讲述出现了数据重复后，该如何处理。

实现去重的通用解决方案：

Kafka 中数据出现重复后，各种解决方案都比较类似，一般需要一个全局 set 集合来维护历史所有数据的主键。当处理新日志时，需要拿到当前日志的主键与历史数据的 set 集合按照规则进行比较，若 set 集合中已经包含了当前日志的主键，说明当前日志在之前已经被处理过了，则当前日志应该被过滤掉，否则认为当前日志不应该被过滤应该被处理，而且处理完成后需要将新日志的主键加入到 set 集合中，set 集合永远存放着所有已经被处理过的数据。

处理流程很简单，关键在于如何维护这个 set 集合，可以简单估算一下这个 set 集合需要占用多大空间。本小节要解决的问题是百亿数据去重，所以就按照每天 1 百亿的数据量来计算。

由于每天数据量巨大，因此主键占用空间通常会比较大，如果主键占用空间小意味着表示的数据范围就比较小，就可能导致主键冲突，例如：4 个字节的 int 类型表示数据范围是为 -2147483648~2147483647，总共可以表示 42 亿个数，如果这里每天百亿的数据量选用 int 类型做为主键的话，很明显会有大量的主键发生冲突，会将不重复的数据认为是发生了重复。

用户的行为日志是在手机客户端生成的，没有全局发号器，一般会选取 UUID 做为日志的主键，UUID 会生成 36 位的字符串，例如："f106c4a1-4c6f-41c1-9d30-bbb2b271284a"。每个主键占用 36 字节，每天 1 百亿数据，36 字节 * 100亿 ≈ 360GB。这仅仅是一天的数据量，所以该 set 集合要想存储空间不发生持续地爆炸式增长，必须增加一个功能，那就是给所有的主键增加 ttl(Time To Live的缩写，即：过期时间)。

如果不增加 ttl，10 天数据量的主键占用空间就 3.6T，100 天数据量的主键占用空间 36T，所以在设计之初必须考虑为主键设定 ttl。如果要求按天进行去重或者认为日志发生重复上报的时间间隔不可能大于 24 小时，那么为了系统的可靠性 ttl 可以设置为 36 小时。每天数据量 1 百亿，且 set 集合中存放着 36 小时的数据量，即 100 亿 * 1.5 = 150 亿，所以 set 集合中需要维护 150 亿的数据量。

且 set 集合中每条数据都增加了 ttl，意味着 set 集合需要为每条数据再附带保存一个时间戳，来确定该数据什么时候过期。例如 Redis 中为一个 key 设置了 ttl，如果没有为这个 key 附带时间戳，那么根本无法判断该 key 什么时候应该被清理。所以在考虑每条数据占用空间时，不仅要考虑数据本身，还需要考虑是否需要其他附带的存储。主键本身占用 36 字节加上 long 类型的时间戳 8 字节，所以每条数据至少需要占用 44 字节，150 亿 * 44 字节 = 660GB。所以每天百亿的数据量，如果我们使用 set 集合的方案来实现，至少需要占用 660GB 以上的存储空间。

使用 BloomFilter 来实现去重

有些流计算的场景对准确性要求并不是很高，例如传统的 Labmda 架构中，都会有离线去矫正实时计算的结果，所以根据业务场景，当业务要求可以接受结果有小量误差时，可以选择使用一些低成本的数据结构。BloomFilter 和 HyperLogLog 都是相对低成本的数据结构，分别有自己的应用场景，且两种数据结构都有一定误差。

HyperLogLog 可以估算出 HyperLogLog 中插入了多少个不重复的元素，而不能告诉我们之前是否插入了哪些元素。BloomFilter 则恰好相反，比起 BloomFilter 更像是一个 set 集合，BloomFilter 可以告诉你 BloomFilter 中肯定不包含元素 a，或者告诉你 BloomFilter 中可能包含元素 b，但 BloomFilter 不能告诉你 BloomFilter 中插入了多少个元素。接下来了解一下 BloomFilter 的实现原理。

bitmap 位图

了解 BloomFilter，从 bitmap(位图)开始说起。现在有 1 千万个整数，数据范围在 0 到 2 千万之间。如何快速查找某个整数是否在这 1 千万个整数中呢？可以将这 1 千万个数保存在 HashMap 中，不考虑对象头及其他空间，1000 万个 int 类型数据需要占用大约 1000 万 * 4 字节 ≈ 40MB 存储空间。有没有其他方案呢？因为数据范围是 0 到 2 千万，所以如下图所示，可以申请一个长度为 2000 万、boolean 类型的数组。将这 1 千万个整数作为数组下标，将其对应的数组值设置成 true，如下图所示，数组下标为 2、666、999 的位置存储的数据为 true，表示 1 千万个数中包含了 2、666、999 等。当查询某个整数 K 是否在这 1 千万个整数中时，只需要将对应的数组值 array[K] 取出来，看是否等于 true。如果等于 true，说明 1 千万整数中包含这个整数 K，否则表示不包含这个整数 K。

Java 的 boolean 基本类型占用一个字节(8bit)的内存空间，所以上述方案需要申请 2000 万字节。如下图所示，可以通过编程语言用二进制位来模拟布尔类型，二进制的 1 表示 true、二进制的 0 表示 false。通过二进制模拟布尔类型的方案，只需要申请 2000 万 bit 即可，相比 boolean 类型而言，存储空间占用仅为原来的 1/8。2000 万 bit ≈ 2.4MB，相比存储原始数据的方案 40 MB 而言，占用的存储空间少了很多。

假如这 1 千万个整数的数据范围是 0 到 100 亿，那么就需要申请 100 亿个 bit 约等于 1200MB，比存储原始数据方案的 40MB 还要大很多。该情况下，直接使用位图使用的存储空间更多了，怎么解决呢？可以只申请 1 亿 bit 的存储空间，对 1000 万个数求hash，映射到 1 亿的二进制位上，最后大约占用 12 MB 的存储空间，但是可能存在 hash 冲突的情况。例如 3 和 100000003(一亿零三)这两个数对一亿求余都为 3，所以映射到长度为 1 亿的位图上，这两个数会占用同一个 bit，就会导致一个问题：1 千万个整数中包含了一亿零三，所以位图中下标为 3 的位置存储着二进制 1。当查询 1 千万个整数中是否包含数字 3 时，同样也是去位图中下标 3 的位置去查找，发现下标为 3 的位置存储着二进制 1，所以误以为 1 千万个整数中包含数字 3。为了减少 hash 冲突，于是诞生了 BloomFilter。

BloomFilter 原理介绍

hash 存在 hash 冲突(碰撞)的问题，两个不同的 key 通过同一个 hash 函数得到的值有可能相同。为了减少冲突，可以多引入几个 hash 函数，如果通过其中的一个 hash 函数发现某元素不在集合中，那么该元素肯定不在集合中。当所有的 hash 函数告诉我们该元素在集合中时，才能确定该元素存在于集合中，这便是BloomFilter的基本思想。

如下图所示，是往 BloomFilter 中插入元素 a、b 的过程，有 3 个 hash 函数，元素 a 经过 3 个 hash 函数后对应的 2、8、10 这三个二进制位，所以将这三个二进制位置为 1，元素 b 经过 3 个 hash 函数后，对应的 5、10、14 这三个二进制位，将这三个二进制位也置为 1，其中下标为 10 的二进制位被 a、b 元素都涉及到。

如下图所示，是从 BloomFilter 中查找元素 c、d 的过程，同样包含了 3 个 hash 函数，元素 c 经过 3 个 hash 函数后对应的 2、6、9 这三个二进制位，其中下标 6 和 9 对应的二进制位为 0，所以会认为 BloomFilter 中不存在元素 c。元素 d 经过 3 个 hash 函数后对应的 5、8、14 这三个二进制位，这三个位对应的二进制位都为 1，所以会认为 BloomFilter 中存在元素 d，但其实 BloomFilter 中并不存在元素 d，是因为元素 a 和元素 b 也对应到了 5、8、14 这三个二进制位上，所以 BloomFilter 会有误判。但是从实现原理来看，当 BloomFilter 告诉你不包含元素 c 时，BloomFilter 中肯定不包含元素 c，当 BloomFilter 告诉你 BloomFilter 中包含元素 d 时，它只是可能包含，也有可能不包含。

使用 BloomFilter 实现数据去重

Redis 4.0 之后 BloomFilter 以插件的形式加入到 Redis 中，关于 api 的具体使用这里不多赘述。BloomFilter 在创建时支持设定一个预期容量和误判率，预期容量即预计插入的数据量，误判率即：当 BloomFilter 中插入的数据达到预期容量时，误判的概率，如果 BloomFilter 中插入数据较少的话，误判率会更低。

经笔者测试，申请一个预期容量为 10 亿，误判率为千分之一的 BloomFilter，BloomFilter 会申请约 143 亿个 bit，即：14G左右，相比之前 660G 的存储空间小太多了。但是在使用过程中，需要记录 BloomFilter 中插入元素的个数，当插入元素个数达到 10 亿时，为了保障误差率，可以将当前 BloomFilter 清除，重新申请一个新的 BloomFilter。

通过使用 Redis 的 BloomFilter，我们可以通过相对较小的内存实现百亿数据的去重，但是 BloomFilter 有误差，所以只能使用在那些对结果能承受一定误差的应用场景，对于广告计费等对数据精度要求非常高的场景，极力推荐大家使用精准去重的方案来实现。

使用 HBase 维护全局 set 实现去重

通过之前分析，我们知道要想实现百亿数据量的精准去重，需要维护 150 亿数据量的 set 集合，每条数据占用 44 KB，总共需要 660 GB 的存储空间。注意这里说的是存储空间而不是内存空间，为什么呢？因为 660G 的内存实在是太贵了，660G 的 Redis 云服务一个月至少要 2 万 RMB 以上，俗话说设计架构不考虑成本等于耍流氓。这里使用 Redis 确实可以解决问题，但是成本较高。HBase 基于 rowkey Get 的效率比较高，所以这里可以考虑将这个大的 set 集合以 HBase rowkey 的形式存放到 HBase 中。HBase 表设置 ttl 为 36 小时，最近 36 小时的 150 亿条日志的主键都存放到 HBase 中，每来一条数据，先拿到主键去 HBase 中查询，如果 HBase 表中存在该主键，说明当前日志已经被处理过了，当前日志应该被过滤。如果 HBase 表中不存在该主键，说明当前日志之前没有被处理过，此时应该被处理，且处理完成后将当前主键 Put 到 HBase 表中。由于数据量比较大，所以一定要提前对 HBase 表进行预分区，将压力分散到各个 RegionServer 上。

使用 HBase rowkey 去重带来的问题

一天 100 亿的数据量，平均一秒 11.57 万条日志。但是数据一般都会有高峰期，例如外卖软件高峰期肯定是饭前的一两个小时，其余时间段数据量相对比较少。所以虽然每天 100 亿数据量，但是到了数据高峰期每秒数据量可以达到 20 万左右。按照之前的思路，每条数据来了都会对 HBase 进行一次 Get 操作，当前数据处理完还会对 HBase 进行一次 Put 操作，所以每秒需要对 HBase 请求 40 万次。单个的 Get 和 Put 请求效率比较低，这里可以优化为批量操作的 API 或异步 API 来提高访问 HBase 的效率。

性能问题优化后，再分析这里使用 HBase 去重到底能不能保证 Exactly Once？拿计算 PV 的案例来讲。

假如 PV 信息维护在 Flink 的状态中，通过幂等性将 PV 统计结果写入到 Redis 供其他业务方查询实时统计的 PV 值。如下图所示，Flink 处理完日志 b 后进行 Checkpoint，将 PV = 2 和 Kafka 对应的 offset 信息保存起来，此时 HBase 表中有两条 rowkey 分别是 a、b，表示主键为 a 和 b 的日志已经被处理过了。

接着往后处理，当处理完日志 d 以后，PV = 4，HBase 表中有 4 条 rowkey 分别是 a、b、c、d，表示主键为 a、b、c、d 的日志已经被处理过了。但此时机器突然故障，导致 Flink 任务挂掉，如右图所示 Flink 任务会从最近一次成功的 Checkpoint 处恢复任务，从日志 b 之后的位置开始消费，且 PV 恢复为 2，因为处理完日志 b 时 PV 为 2。

但由于 HBase 中的数据不是由 Flink 来维护，所以无法恢复到 Checkpoint 时的状态。所以 Flink 任务恢复后，PV = 2 且 HBase 中 rowkey 为 a、b、c、d。此时 Flink 任务从日志 c 开始继续处理数据，当处理日志 c 和 d 时，Flink 任务会先查询 HBase，发现 HBase 中已经保存了主键 c 和 d，所以认为日志 c 和 d 已经被处理了，会将日志 c 和 d 过滤掉，于是就产生了丢数据的现象，日志 c 和 d 其实并没有参与 PV 的计算。

日志 c 和 d 已经被处理过了，此时就算从 Checkpoint 处恢复，PV 值也应该为 4，不应该是 2。请注意上述方案，笔者描述的是 PV 信息维护在 Flink 的状态中，所以从 Checkpoint 处恢复任务时，会将 Checkpoint 时状态中保存的 PV 信息恢复，所以恢复为 2。

当然还有其他统计 PV 的方式，不需要将 PV 信息维护在 Flink 状态中，而是仅仅在 Redis 中保存 PV 结果，每处理一条数据，将 Redis 中的 PV 值加一即可。如下图所示，PV 不维护在状态中，所以当处理完日志 b 进行 Checkpoint 时，只会将当前消费的 offset 信息维护起来。处理完日志 d 以后，由于机器故障，Flink 任务挂掉，任务依然会从日志 b 之后开始消费，此时 Redis 中保存的 PV=4，且 HBase 中保存的 rowkey 信息为 a、b、c、d。紧接着开始处理 c 和 d，因为 HBase 中保存了主键 c、d，因此不会重复处理日志 c、d，因此 PV 值计算正确，也不会出现重复消费的问题。

这种策略貌似没有问题，但是问题百出。我们的任务处理元素 d 需要两个操作：

① 将 Redis 中 PV 值加一 ② 将主键 id 加入到 HBase

由于 Redis 和 HBase 都不支持事务，所以以上两个操作并不能保障原子性。如果代码中先执行步骤 ①，可能会造成 ① 执行成功 ② 还未执行成功，那么恢复任务时 PV=4，HBase 中保存主键 a、b、c，此时日志 d 就会重复计算，就会造成 PV 值计算偏高的问题。如果代码中先执行步骤 ②，可能会造成 ② 执行成功 ① 还未执行成功，那么恢复任务时 PV=3，HBase 中保存主键 a、b、c、d，此时日志 d 就会被漏计算，就会造成 PV 值计算偏低的问题。这里只是拿 HBase 举例而已，上述情况中外部的任何存储介质维护 set 集合都不能保证 Exactly Once，因为 Flink 从 Checkpoint 处恢复时，外部存储介质并不能恢复到 Checkpoint 时的状态。既然外部存储介质不能恢复到 Checkpoint 时的状态，那使用 Flink 内置的状态后端可以吗？当然可以！！！

使用 Flink 的 KeyedState 实现去重

使用 Flink 状态来维护 set 集合的优势