每日算法 — 使用java实现HyperLogLog：基数估计与大数据近似去重计数

算法概述

HyperLogLog 是一种概率型数据结构，用于在极大集合中高效估计不同元素的数量（基数）。其核心思想基于观察随机比特流的规律：若将元素通过哈希函数映射为均匀分布的比特串，则出现连续 k 个前导零的概率约为 1/2^k。通过记录多个桶中的最大前导零长度，HyperLogLog 能以极小的内存开销（通常几KB）提供基数估计，标准误差约 1.04/√m（m 为桶数）。

本文用 Java 实现一个标准 HyperLogLog，支持 add、count 和 merge 操作，适用于海量数据的去重计数场景。

核心原理

1. 哈希映射：将输入元素通过 64 位哈希函数映射为伪随机比特串。
2. 桶分配：取哈希值的高若干位作为桶索引，剩余低位用于计算前导零长度。
3. 前导零统计：从哈希低位开始统计前导零个数，记录每个桶的最大值。
4. 调和均值估计：使用桶中最大前导零长度的调和均值作为基数估计的基础。
5. 偏差修正：对小基数和大基数分别应用经验修正公式。

Java 实现

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;

/**
 * HyperLogLog 基数估计算法实现
 * 标准误差约 1.04 / sqrt(2^p)，其中 p 为精度参数
 */
public class HyperLogLog {
    // 精度参数 p，决定桶的数量 m = 2^p
    private final int p;
    // 桶的数量
    private final int m;
    // 每个桶记录的最大前导零长度（加1偏移）
    private final byte[] registers;
    // 用于计算 alpha 常数
    private final double alphaMM;

    /**
     * 构造函数
     * @param p 精度参数，通常取 4~16，推荐 14（16K桶，误差约 0.81%）
     */
    public HyperLogLog(int p) {
        if (p < 4 || p > 16) {
            throw new IllegalArgumentException("精度参数 p 必须在 4~16 之间");
        }
        this.p = p;
        this.m = 1 << p; // 2^p
        this.registers = new byte[m];
        this.alphaMM = getAlphaM(m) * m * m;
    }

    /**
     * 获取 alpha 修正系数
     */
    private double getAlphaM(int m) {
        switch (m) {
            case 16: return 0.673;
            case 32: return 0.697;
            case 64: return 0.709;
            default: return 0.7213 / (1 + 1.079 / m);
        }
    }

    /**
     * 向 HyperLogLog 中添加一个元素
     * @param element 待添加的字符串元素
     */
    public void add(String element) {
        long hash = murmurHash64(element);
        // 高 p 位作为桶索引
        int bucket = (int) (hash >>> (64 - p));
        // 剩余低位计算前导零长度
        long remaining = hash << p; // 左移去掉高p位
        int leadingZeros = numberOfLeadingZeros(remaining) + 1;
        // 更新桶中的最大值
        if (leadingZeros > registers[bucket]) {
            registers[bucket] = (byte) leadingZeros;
        }
    }

    /**
     * 计算 64 位长整型的前导零个数
     */
    private int numberOfLeadingZeros(long x) {
        if (x == 0) return 64;
        int n = 1;
        int y = (int) (x >>> 32);
        if (y == 0) { n += 32; y = (int) x; }
        if (y >>> 16 == 0) { n += 16; y <<= 16; }
        if (y >>> 24 == 0) { n += 8; y <<= 8; }
        if (y >>> 28 == 0) { n += 4; y <<= 4; }
        if (y >>> 30 == 0) { n += 2; y <<= 2; }
        n -= y >>> 31;
        return n;
    }

    /**
     * MurmurHash3 64位变体，提供均匀分布的哈希值
     */
    private long murmurHash64(String key) {
        byte[] data = key.getBytes(StandardCharsets.UTF_8);
        long seed = 0x123456789ABCDEF0L;
        long m = 0xc6a4a7935bd1e995L;
        int r = 47;
        long h = seed ^ (data.length * m);
        int len = data.length;
        int len8 = len / 8;

        for (int i = 0; i < len8; i++) {
            int idx = i * 8;
            long k = ((long) data[idx] & 0xff)
                    | (((long) data[idx + 1] & 0xff) << 8)
                    | (((long) data[idx + 2] & 0xff) << 16)
                    | (((long) data[idx + 3] & 0xff) << 24)
                    | (((long) data[idx + 4] & 0xff) << 32)
                    | (((long) data[idx + 5] & 0xff) << 40)
                    | (((long) data[idx + 6] & 0xff) << 48)
                    | (((long) data[idx + 7] & 0xff) << 56);
            k *= m;
            k ^= k >>> r;
            k *= m;
            h ^= k;
            h *= m;
        }

        // 处理剩余字节
        switch (len % 8) {
            case 7: h ^= (long) (data[len8 * 8 + 6] & 0xff) << 48;
            case 6: h ^= (long) (data[len8 * 8 + 5] & 0xff) << 40;
            case 5: h ^= (long) (data[len8 * 8 + 4] & 0xff) << 32;
            case 4: h ^= (long) (data[len8 * 8 + 3] & 0xff) << 24;
            case 3: h ^= (long) (data[len8 * 8 + 2] & 0xff) << 16;
            case 2: h ^= (long) (data[len8 * 8 + 1] & 0xff) << 8;
            case 1: h ^= (long) (data[len8 * 8] & 0xff);
                    h *= m;
        }

        h ^= h >>> r;
        h *= m;
        h ^= h >>> r;
        return h;
    }

    /**
     * 估计当前集合的基数（不同元素数量）
     * @return 基数估计值
     */
    public long count() {
        double rawEstimate = alphaMM / harmonicMean();

        // 小基数修正：使用线性计数
        if (rawEstimate <= 2.5 * m) {
            int zeros = countZeros();
            if (zeros != 0) {
                return Math.round(m * Math.log((double) m / zeros));
            }
        }
        // 大基数修正（超过 2^32 时使用 64 位修正）
        // 对于 64 位哈希，此实现暂不做额外修正
        return Math.round(rawEstimate);
    }

    /**
     * 计算调和均值
     */
    private double harmonicMean() {
        double sum = 0.0;
        for (byte reg : registers) {
            sum += 1.0 / (1L << reg); // 2^(-register)
        }
        return sum;
    }

    /**
     * 统计值为 0 的桶数量
     */
    private int countZeros() {
        int count = 0;
        for (byte reg : registers) {
            if (reg == 0) count++;
        }
        return count;
    }

    /**
     * 合并另一个 HyperLogLog（要求相同精度）
     * @param other 另一个 HyperLogLog 实例
     */
    public void merge(HyperLogLog other) {
        if (this.p != other.p) {
            throw new IllegalArgumentException("只能合并相同精度的 HyperLogLog");
        }
        for (int i = 0; i < m; i++) {
            if (other.registers[i] > this.registers[i]) {
                this.registers[i] = other.registers[i];
            }
        }
    }

    @Override
    public String toString() {
        return "HyperLogLog(p=" + p + ", m=" + m + ", estimate=" + count() + ")";
    }

    // ==================== 测试入口 ====================
    public static void main(String[] args) {
        HyperLogLog hll = new HyperLogLog(14); // 16384 桶

        // 模拟添加 100 万个不同的用户 ID
        int actualUnique = 1_000_000;
        for (int i = 0; i < actualUnique; i++) {
            hll.add("user_" + i);
        }

        long estimate = hll.count();
        double error = Math.abs(estimate - actualUnique) * 100.0 / actualUnique;

        System.out.println("实际基数: " + actualUnique);
        System.out.println("估计基数: " + estimate);
        System.out.printf("相对误差: %.4f%%\n", error);

        // 测试合并功能
        HyperLogLog hll2 = new HyperLogLog(14);
        for (int i = 500_000; i < 1_500_000; i++) {
            hll2.add("user_" + i);
        }
        hll.merge(hll2);
        System.out.println("合并后估计: " + hll.count());
    }
}

复杂度分析

其中 m = 2^p 为桶数。当 p = 14 时，空间仅需 16KB 左右，却能估计数十亿级别的基数。

适用场景

• 网站 UV 统计（海量独立访客去重）
• 分布式系统中的全局去重计数
• 实时流数据中的不同值数量估计
• 数据库查询优化中的基数估计