算法 也就是基数估计统计算法，预估一个集合中不同数据的个数，也就是我们常说的去重统计java 去重算法，在redis中也存在 类型的结构java 去重算法 Flink去重第三弹：HyperLogLog去重，能够使用12k的内存，允许误差在0.81%的情况下统计2^64个数据，在这种大数据量情况下能够减少存储空间的消耗，但是前提是允许存在一定的误差。关于算法原理可以参考这篇文章：探索算法（含Java实现）里面做了详细的介绍，其算法实现在开源java流式计算库-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 values =new HashSet();
HyperLogLog logLog=new HyperLogLog(0.01); //允许误差
String line = "";
while ((line = br.readLine()) != null) {
String[] s = line.split(",");
String uuid = s[0];
<p>![java hash算法原理_java 去重算法_java数据结构和算法][1]
values.add(uuid);
logLog.offer(uuid);
}
long rs=logLog.cardinality();</p>
当误差值为0.01 时; rs为98228，需要内存大小int[1366] //内部数据结构
当误差值为0.001时；rs为97304 ，需要内存大小int[]
误差越小也就越来越接近其真实数据java 去重算法，但是在这个过程中需要的内存也就越来越大，这个取舍可根据实际情况决定。
在开发中更多希望通过sql方式来完成，那么就将hll与udaf结合起来使用，实现代码如下：
public class HLLDistinctFunction extends AggregateFunction {
@Override public HyperLogLog createAccumulator() {
<p>
return new HyperLogLog(0.001);
}
public void accumulate(HyperLogLog hll,String id){
hll.offer(id);
}
@Override public Long getValue(HyperLogLog accumulator) {
return accumulator.cardinality();
}</p>
定义的返回类型是long 也就是去重的结果，是一个类型的结构。
测试：
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=>{
<p>![java hash算法原理_java数据结构和算法_java 去重算法][3]
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()
}</p>
准备测试数据
`
1,,00
1,,00
1,,00
`
得到结果：
`
4> (true,1,00)
4> (false,1,00)
4> (true,2,00)
`
其基本使用介绍到这里，后续还将进一步优化。
关注回复Flink获取更多系列文章