Flink去重第三弹：HyperLogLog去重

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,15778080000001,devId2,15778080000001,devId1,1577808000000

得到结果：

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

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

原文地址：https://www.cnblogs.com/pucheung/p/12184785.html