Flink去重第三弹：HyperLogLog去重

向右看齐 2023-06-27 06:27 65阅读 0赞

format_png

点击箭头处“蓝色字”，关注我们哦！！

HyperLogLog算法 也就是基数估计统计算法，预估一个集合中不同数据的个数，也就是我们常说的去重统计，在redis中也存在hyperloglog 类型的结构，能够使用12k的内存，允许误差在0.81%的情况下统计2^64个数据，在这种大数据量情况下能够减少存储空间的消耗，但是前提是允许存在一定的误差。关于HyperLogLog算法原理可以参考这篇文章：https://www.jianshu.com/p/55defda6dcd2里面做了详细的介绍，其算法实现在开源java流式计算库stream-lib提供了其具体实现代码，由于代码比较长就不贴出来(可以后台回复hll ,获取flink使用hll去重的完整代码)。

测试一下其使用效果，准备了97320不同数据：

  1. public static void main(String[] args) throws Exception{
  3.         String filePath = "000000_0";
  4.         BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
  6.         Set<String> values =new HashSet<>();
  7.         HyperLogLog logLog=new HyperLogLog(0.01); //允许误差
  9.         String line = "";
  10.         while ((line = br.readLine()) != null) {
  11.             String[] s = line.split(",");
  12.             String uuid = s[0];
  13.             values.add(uuid);
  14.             logLog.offer(uuid);
  15.         }
  17.         long rs=logLog.cardinality();
  18.     }

当误差值为0.01 时; rs为98228，需要内存大小int[1366] //内部数据结构
当误差值为0.001时；rs为97304 ，需要内存大小int[174763]
误差越小也就越来越接近其真实数据，但是在这个过程中需要的内存也就越来越大，这个取舍可根据实际情况决定。

在开发中更多希望通过sql方式来完成，那么就将hll与udaf结合起来使用，实现代码如下：

  1. public class HLLDistinctFunction extends AggregateFunction<Long,HyperLogLog> {
  3.     @Override public HyperLogLog createAccumulator() {
  4.         return new HyperLogLog(0.001);
  5.     }
  7.     public void accumulate(HyperLogLog hll,String id){
  8.       hll.offer(id);
  9.     }
  11.     @Override public Long getValue(HyperLogLog accumulator) {
  12.         return accumulator.cardinality();
  13.     }
  14. }

定义的返回类型是long 也就是去重的结果，accumulator是一个HyperLogLog类型的结构。

测试：

  1. case class AdData(id:Int,devId:String,datatime:Long)object Distinct1 {  def main(args: Array[String]): Unit = {
  2.     val env=StreamExecutionEnvironment.getExecutionEnvironment
  3.     val tabEnv=StreamTableEnvironment.create(env)
  4.     tabEnv.registerFunction("hllDistinct",new HLLDistinctFunction)
  5.     val kafkaConfig=new Properties()
  6.    kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092")
  7.     kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"test1")
  8.     val consumer=new FlinkKafkaConsumer[String]("topic1",new SimpleStringSchema,kafkaConfig)
  9.     consumer.setStartFromLatest()
  10.     val ds=env.addSource(consumer)
  11.       .map(x=>{
  12.         val s=x.split(",")
  13.         AdData(s(0).toInt,s(1),s(2).toLong)
  14.       })
  15.     tabEnv.registerDataStream("pv",ds)
  16.     val rs=tabEnv.sqlQuery(      """ select hllDistinct(devId) ,datatime
  17.                                           from pv group by datatime
  18.       """.stripMargin)
  19.     rs.writeToSink(new PaulRetractStreamTableSink)
  20.     env.execute()
  21.   }
  22. }

准备测试数据

  1. 1,devId1,1577808000000
  2. 1,devId2,1577808000000
  3. 1,devId1,1577808000000

得到结果：

  1. 4> (true,1,1577808000000)
  2. 4> (false,1,1577808000000)
  3. 4> (true,2,1577808000000)

其基本使用介绍到这里，后续还将进一步优化。

—END—

format_png 1

关注回复Flink

获取更多系列

好看，就要点个”在看”

format_png 2

发表评论

表情：
评论列表 （有 0 条评论，65人围观）

还没有评论，来说两句吧...

相关阅读

    相关 flink sql去重

    消沉了一段时间，但是生活还是要继续的。 一直以来收集的数据都有重复的情况，flink sql如何去重呢？ 1、 distinct统计 对devId去重

    Dear 丶Dear 丶/ 0 赞/ 15 阅读

    相关 Flink去重第一弹:MapState去重

    点击上方蓝 字关注~         去重计算应该是数据分析业务里面常见的指标计算，例如网站一天的访问用户数、广告的点击用户数等等，离线计算是一个全量、一次性计算的过程通

    水深无声水深无声/ 0 赞/ 91 阅读