stream-lib流式计算库

目录

简介

HyperLogLog概述

maven依赖

测试用例

参考

---

简介

stream-lib是一个开源的Java流式计算库，里面有很多大数据估值算法的实现，也包括HyperLogLog算法。

HyperLogLog概述

HyperLogLog 会计算每一个成员二进制值首位有多少个零，如果零的最大个数是 n，则唯一值数量就是 2^n。算法中有两个关键点，首先，成员的值必须是服从正态分布的，这一点可以通过哈希函数实现。stream-lib 使用的是 MurmurHash，它简单、快速、且符合分布要求，应用于多种基于哈希查询的算法。其次，为了降低计算结果的方差，集合成员会先被拆分成多个子集合，最后的唯一值数量是各个子集合结果的调和平均数。上文代码中，我们传递给 HyperLogLog构造函数的整型参数就表示会采用多少个二进制位来进行分桶。最后，准确性可以通过这个公式计算：1.04/sqrt(2^log2m)。

HyperLogLog 是对 LogLog 算法的扩展，而 HyperLogLogPlus 则包含了更多优化策略。比如，它使用了 64 位的哈希函数，以减少哈希碰撞；对于唯一值数较小的集合，会引入纠偏机制；此外，它还对子集合的存储方式做了改进，能够从稀疏型的数据结构逐渐扩展为密集型。

maven依赖

<dependency>
    <groupId>com.clearspring.analytics</groupId>
    <artifactId>stream</artifactId>
    <version>2.5.2</version>
</dependency>

测试用例

import com.clearspring.analytics.stream.cardinality.CardinalityMergeException;
import com.clearspring.analytics.stream.cardinality.HyperLogLog;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class HyperLogLogDemo {
    public static void main(String[] args) throws IOException, CardinalityMergeException {
        HyperLogLog hyperLogLog1 = new HyperLogLog(14);
        for (int i = 1;i < 100000000;i++) {
            hyperLogLog1.offer(i); // 插入数据
        }
        HyperLogLog hyperLogLog2 = new HyperLogLog(14);
        for (int i = 50000;i < 200000;i++) {
            hyperLogLog2.offer(i);
        }

        // 序列化成字符串
        String hllStr1 = new String(hyperLogLog1.getBytes(), StandardCharsets.ISO_8859_1);
        String hllStr2 = new String(hyperLogLog2.getBytes(), StandardCharsets.ISO_8859_1);

        // 反序列化
        HyperLogLog hll1 = null;
        long start = System.currentTimeMillis();
        for (int i = 0; i < 50000; i++) {
            hll1 = HyperLogLog.Builder.build(hllStr1.getBytes(StandardCharsets.ISO_8859_1));
        }
        System.out.println("反序列化耗时：" + (System.currentTimeMillis() - start));

        HyperLogLog hll2 = HyperLogLog.Builder.build(hllStr2.getBytes(StandardCharsets.ISO_8859_1));

        start = System.currentTimeMillis();
        for (int i = 0; i < 50000; i++) {
            // 合并
            hll1.addAll(hll2);
        }

        System.out.println("合并耗时：" + (System.currentTimeMillis() - start));

        // 输出去重后的统计数
        System.out.println(hll1.cardinality());
    }
}

参考

github地址：https://github.com/addthis/stream-lib

hyperLogLog算法介绍：https://www.jianshu.com/p/55defda6dcd2