Flink去重第三弹：HyperLogLog去重

Posted 2020-11-24 pucheung

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)

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