亿万级海量数据去重软方法，spark/hive/flink/mr通用

一、场景描述：

​ 小强作为一名数据工程师，给予hadoop生态，经常会接到类似uv的去重统计。对于这种需求，一般的数据工程师撸起袖子直接干！一般情况下不会有问题。某一天，你公司突然业务发展发展起来，数据量慢慢暴涨，你会突然发现之前的count distinct去重经常oom或是龟速出数据。上来一股脑加内存！加！果断加！某一天你老板要你在原来按天的uv加一个月uv、年uv，这时你慌了。只会说“老板！加机器，内存不够！”。老板说：“算个uv你就想骗我钱？你明天不用来上班了！”

​ 打不死的小强这时拼命百度，在网上找到许多神乎其神的方法…

二、常用方法

1.优化sql

​ 小强把原有的count distinct去重改成了group by，性能也提升了不少。安稳的日子过了一段后，公司数据量也一点一点增长，小强上来一把hive SQL，成功扛住：

select count(*) from (select uid,pid as upv from table_event group by uid,pid) tmp

老板突然说要在某某维度下，加个upv统计，xxx的统计，指标不断增加，这时，小强的sql不得不改变：

select collect_set(uid) as uv,collect_set(uid,pid) as upv from table_event group by xxx,xx,x

小强一执行，直接oom，于是，加内存，加内存，最后终于跑成功了，这时，集群内存也耗尽，upv十亿级，加上某些异常值的倾斜严重！小强不服气，又喊:“老板！加资源！没内存了！”。老板：“现在疫情期间，公司财务紧张，你做不了就走吧，没赔偿！”

2.借助第三方存储

小强不想被离职，想被老板认可，不甘被这种小需求难道！不停探索。想到一个法子：利用外部K-V数据库（Redis、HBase之类）存储需要去重的键，最后统计一把键的数量即可。做着做着，小强被这三点打回：

• 外部存储介质不熟悉，维护成本大
• 取值方式与现有方式差别太大，需要单独处理
• 操作麻烦，需要写单独的udf

重重困难后，小强又放弃了…

3.bitmap

最后小强找到一种高大上的方法，老板都没听过，准备把这个方法拿去忽悠老板，结果老板没忽悠到，自己被下面几个难倒了

• 侵入性太大，需要引入外部依赖，与现有环境冲突
• 需要维护bit位，需要想办法把要去重的字符型id转为int或long类型
• 扩展性太差，再加个维度需要重新编码bit位

太难了。。。最后。。。。小强被开除了。。。就是一个这么悲伤的故事

三、原理分析

在大数据分布式计算框架生态下，提升计算效率的方法是尽可能的把计算分布式话、并行化，避免单节点计算过载，把计算分摊到各个节点。这样解释小白能够听懂：比如你有5个桶，怎样轻松地把A池子的水倒入B池子里？

• 最大并行化，5个桶同时利用，避免count distinct只用一个桶的方法
• 重复利用化，一次提不动那么多水，不要打肿脸充胖子，一不小心oom，为什么不分几次呢
• 数据均衡化，5个桶的水不要一个多一个少的，第一个提水的次数变多，第二个某些桶扛不住，俗称数据倾斜

四、案例实战

通过案例来说明海量数据如何高效的去重，下面是原始数据，要计算day_num维度下的uv，自己脑补出海量数据，这里为方便说明，只列举了day_num，一个维度用桶来描绘计算模型，假设数据都是按字典顺序分桶

> select * from event;
+----------------+------------+
| event.day_num  | event.uid  |
+----------------+------------+
| day1           | a          |
| day1           | a          |
| day1           | a          |
| day1           | a          |
| day1           | bb         |
| day1           | bb         |
| day1           | bbb        |
| day1           | ccc        |
| day1           | ccc        |
| day1           | dddd       |
| day1           | eeee       |
| day1           | eeeee      |
| day1           | eeeee      |
| day1           | eeeee      |
+----------------+------------+

• 原始方法，使用count distinct

select count(distinct(uid))as uv from event group by day_num;

桶的使用如下：

可以看到所有数据装到一个桶里面，桶已经快装不下了，明显最差

• 优化方法1

select size(collect_set(uid)) as uv from (select day_num,uid from event group by day_num,uid) tmp group by day_num;

桶的使用如下：

可以明显看到比上面方法有进步，充分利用了桶，最大的实现了并行化，执行虽然分为了两部，但是大大减轻了第一步的负担，面向海量数据的场景去重方面拥有绝对的优势，假如第二步的结果集还是太大了呢？一样会oom扛不住

• 优化方法2（推荐）

简单说就是转化计算，在一个jvm里面，硬去重的方法都逃不开把所有字符或字符的映射放一个对象里面，通过一定的逻辑获取去重集合，对于分布式海量数据的场景下，这种硬去重的计算仍然会花大量的时间在上图的最后单点去重的步骤，我们可以把去重的逻辑按照一定的规则分桶计算完成，每个桶之间分的数据都不重复，所有桶计算完桶内数据去重的集合大小，最后一步再相加。讲的有点抽象，上代码

create table event_tmp as select *,length(uid) as len_uid from event;

为了方便说明，我拆分步骤，创建临时表，其中length(uid) as len_uid是映射字段，uid的长度

select sum(uv_tmp) as uv 
from
  (
      select day_num,size(collect_set(uid)) as uv_tmp 
       from event_tmp 
       group by len_uid,day_num
  ) tmp group by day_num

桶的使用如下：

这里使用uid长度映射字段，实际开发中，你也可以选择首字母、末字母或者其它能想到的属性作为映射字段，分桶分步预聚合的方法，巧妙的把一个集合去重问题最终转化为相加问题，避开了单个jvm去重承受的压力，在海量数据的场景下，这个方法最为使用，推荐用在生产上。

五、总结

海量数据高效去重的思想就是最大的把计算和数据并行化，充分利用、均衡利用分布式集群下的算力，避开单点压力，强去重的方法在小数据量下会有优势，在海量数据下去重，必须要考虑转换思想。上面的优化方法，举了个简单的栗子，在实际开发当中，不仅仅是sql，在编写spark flink程序里面思想也一样通用，尤其是实时去重，用强去重的方法你得始终维护一个大集合，这样会代码很大的资源浪费和维护成本，想办法把要去重的数据映射一个可以均分数据key出来做预聚合，别来硬的，试试软方法