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,1577808000000
1,devId2,1577808000000
1,devId1,1577808000000
得到结果：
4> (true,1,1577808000000)
4> (false,1,1577808000000)
4> (true,2,1577808000000)
其基本使用介绍到这里，后续还将进一步优化。
—END—
本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2020-01-04，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
相关产品与服务
流计算 Oceanus
流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的企业级实时大数据分析平台，具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。
腾讯云开发者
