Flink去重第三弹：HyperLogLog去重

向右看齐 2023-06-27 06:27 10阅读 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不同数据：

public static void main(String[] args) throws Exception{
    
            String filePath = "000000_0";
            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
    
            Set<String> values =new HashSet<>();
            HyperLogLog logLog=new HyperLogLog(0.01); //允许误差
    
            String line = "";
            while ((line = br.readLine()) != null) {
                String[] s = line.split(",");
                String uuid = s[0];
                values.add(uuid);
                logLog.offer(uuid);
            }
           
            long rs=logLog.cardinality();
        }

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

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

public class HLLDistinctFunction extends AggregateFunction<Long,HyperLogLog> {
    
        @Override public HyperLogLog createAccumulator() {
            return new HyperLogLog(0.001);
        }
    
        public void accumulate(HyperLogLog hll,String id){
          hll.offer(id);
        }
    
        @Override public Long getValue(HyperLogLog accumulator) {
            return accumulator.cardinality();
        }
    }

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

测试：

case class AdData(id:Int,devId:String,datatime:Long)object Distinct1 {  def main(args: Array[String]): Unit = {
        val env=StreamExecutionEnvironment.getExecutionEnvironment
        val tabEnv=StreamTableEnvironment.create(env)
        tabEnv.registerFunction("hllDistinct",new HLLDistinctFunction)
        val kafkaConfig=new Properties()
       kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092")
        kafkaConfig.put(ConsumerConfig.GROUP_ID_CONFIG,"test1")
        val consumer=new FlinkKafkaConsumer[String]("topic1",new SimpleStringSchema,kafkaConfig)
        consumer.setStartFromLatest()
        val ds=env.addSource(consumer)
          .map(x=>{
            val s=x.split(",")
            AdData(s(0).toInt,s(1),s(2).toLong)
          })
        tabEnv.registerDataStream("pv",ds)
        val rs=tabEnv.sqlQuery(      """ select hllDistinct(devId) ,datatime
                                              from pv group by datatime
          """.stripMargin)
        rs.writeToSink(new PaulRetractStreamTableSink)
        env.execute()
      }
    }

准备测试数据

1,devId1,1577808000000
    1,devId2,1577808000000
    1,devId1,1577808000000

得到结果：

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

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

**—END—**

![format_png 1][]

关注回复Flink

获取更多系列

好看，就要点个"在看"

![format_png 2][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6L1VDNk0xSGYwU1NFcHIxbGJiaWF0aWF4UEpjOHk5SmNPZVJ5Sk9JaHNpYndTeFdQbUhDcUp3eldYOHhUTXo5TVlMcEhLQWtSZlkyZmNNcW92eXhyUjlLcEF3LzY0MA?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9TN3RJdm81eFlNNjNhR0FoNlZmT0ZJMHBDVnFrVXRkbkJXWUYzVkFaOUVLaGNpYzhWeEtWT1JpY2MzQVFiN2d2ZkVOT1FnQ1dycFZoc2lhdXFhanR0Unc3US82NDA?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2dpZi9TN3RJdm81eFlNN01GRVJScTAzTTkwQUVIdTBJRUFSOFpEM21BVzhyZDRJQUZrSGxndEZnaWNCNll1Z014bXdCMk1TVmIyWllMc1kyaWJpYU9pYVdsNHVjUncvNjQw?x-oss-process=image/format,png