伍佰目录 短网址
  当前位置:海洋目录网 » 站长资讯 » 站长资讯 » 文章详细 订阅RssFeed

Flink去重第三弹:HyperLogLog去重

来源:本站原创 浏览:86次 时间:2023-05-10
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)

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


  推荐站点

  • At-lib分类目录At-lib分类目录

    At-lib网站分类目录汇集全国所有高质量网站,是中国权威的中文网站分类目录,给站长提供免费网址目录提交收录和推荐最新最全的优秀网站大全是名站导航之家

    www.at-lib.cn
  • 中国链接目录中国链接目录

    中国链接目录简称链接目录,是收录优秀网站和淘宝网店的网站分类目录,为您提供优质的网址导航服务,也是网店进行收录推广,站长免费推广网站、加快百度收录、增加友情链接和网站外链的平台。

    www.cnlink.org
  • 35目录网35目录网

    35目录免费收录各类优秀网站,全力打造互动式网站目录,提供网站分类目录检索,关键字搜索功能。欢迎您向35目录推荐、提交优秀网站。

    www.35mulu.com
  • 就要爱网站目录就要爱网站目录

    就要爱网站目录,按主题和类别列出网站。所有提交的网站都经过人工审查,确保质量和无垃圾邮件的结果。

    www.912219.com
  • 伍佰目录伍佰目录

    伍佰网站目录免费收录各类优秀网站,全力打造互动式网站目录,提供网站分类目录检索,关键字搜索功能。欢迎您向伍佰目录推荐、提交优秀网站。

    www.wbwb.net