探索HyperLogLog算法（含Java实现）(3)

如果理解了之前的分桶算法，那么很显然分桶数只能是2的整数次幂。
如果分桶越多，那么估计的精度就会越高，统计学上用来衡量估计精度的一个指标是“相对标准误差”(relative standard deviation，简称RSD)，RSD的计算公式这里就不给出了，百科上一搜就可以知道，从直观上理解，RSD的值其实就是（(每次估计的值）在（估计均值）上下的波动）占（估计均值）的比例（这句话加那么多括号是为了方便大家断句）。RSD的值与分桶数m存在如下的计算关系：

RSD

有了这个公式，你可以先确定你想要达到的RSD的值，然后再推出分桶的数目m。

Java实现分析

stream-lib是一个开源的Java流式计算库，里面有很多大数据估值算法的实现，其中当然包括HyperLogLog算法，HyperLogLog实现类的代码地址如下：
https://github.com/addthis/stream-lib/blob/master/src/main/java/com/clearspring/analytics/stream/cardinality/HyperLogLog.java

这个实现类中还包含很多与算法无关的序列化之类的代码，所以不建议你直接去看，我把它的算法主干抽取了出来，变成了如下的三个类，你把这三个类的代码复制下来放到项目的同一个包下即可，HyperLogLog类中还包含一个main函数，你可以运行一下看看代码是否正确，代码如下：
HyperLogLog.java

public class HyperLogLog { private final RegisterSet registerSet; private final int log2m; //log(m) private final double alphaMM; /** * * rsd = 1.04/sqrt(m) * @param rsd 相对标准偏差 */ public HyperLogLog(double rsd) { this(log2m(rsd)); } /** * rsd = 1.04/sqrt(m) * m = (1.04 / rsd)^2 * @param rsd 相对标准偏差 * @return */ private static int log2m(double rsd) { return (int) (Math.log((1.106 / rsd) * (1.106 / rsd)) / Math.log(2)); } private static double rsd(int log2m) { return 1.106 / Math.sqrt(Math.exp(log2m * Math.log(2))); } /** * accuracy = 1.04/sqrt(2^log2m) * * @param log2m */ public HyperLogLog(int log2m) { this(log2m, new RegisterSet(1 << log2m)); } /** * * @param registerSet */ public HyperLogLog(int log2m, RegisterSet registerSet) { this.registerSet = registerSet; this.log2m = log2m; int m = 1 << this.log2m; //从log2m中算出m alphaMM = getAlphaMM(log2m, m); } public boolean offerHashed(int hashedValue) { // j 代表第几个桶,取hashedValue的前log2m位即可 // j 介于 0 到 m final int j = hashedValue >>> (Integer.SIZE - log2m); // r代表 除去前log2m位剩下部分的前导零 + 1 final int r = Integer.numberOfLeadingZeros((hashedValue << this.log2m) | (1 << (this.log2m - 1)) + 1) + 1; return registerSet.updateIfGreater(j, r); } /** * 添加元素 * @param o 要被添加的元素 * @return */ public boolean offer(Object o) { final int x = MurmurHash.hash(o); return offerHashed(x); } public long cardinality() { double registerSum = 0; int count = registerSet.count; double zeros = 0.0; //count是桶的数量 for (int j = 0; j < registerSet.count; j++) { int val = registerSet.get(j); registerSum += 1.0 / (1 << val); if (val == 0) { zeros++; } } double estimate = alphaMM * (1 / registerSum); if (estimate <= (5.0 / 2.0) * count) { //小数据量修正 return Math.round(linearCounting(count, zeros)); } else { return Math.round(estimate); } } /** * 计算constant常数的取值 * @param p log2m * @param m m * @return */ protected static double getAlphaMM(final int p, final int m) { // See the paper. switch (p) { case 4: return 0.673 * m * m; case 5: return 0.697 * m * m; case 6: return 0.709 * m * m; default: return (0.7213 / (1 + 1.079 / m)) * m * m; } } /** * * @param m 桶的数目 * @param V 桶中0的数目 * @return */ protected static double linearCounting(int m, double V) { return m * Math.log(m / V); } public static void main(String[] args) { HyperLogLog hyperLogLog = new HyperLogLog(0.1325);//64个桶 //集合中只有下面这些元素 hyperLogLog.offer("hhh"); hyperLogLog.offer("mmm"); hyperLogLog.offer("ccc"); //估算基数 System.out.println(hyperLogLog.cardinality()); } }

MurmurHash.java

/** * 一种快速的非加密hash * 适用于对保密性要求不高以及不在意hash碰撞攻击的场合 */ public class MurmurHash { public static int hash(Object o) { if (o == null) { return 0; } if (o instanceof Long) { return hashLong((Long) o); } if (o instanceof Integer) { return hashLong((Integer) o); } if (o instanceof Double) { return hashLong(Double.doubleToRawLongBits((Double) o)); } if (o instanceof Float) { return hashLong(Float.floatToRawIntBits((Float) o)); } if (o instanceof String) { return hash(((String) o).getBytes()); } if (o instanceof byte[]) { return hash((byte[]) o); } return hash(o.toString()); } public static int hash(byte[] data) { return hash(data, data.length, -1); } public static int hash(byte[] data, int seed) { return hash(data, data.length, seed); } public static int hash(byte[] data, int length, int seed) { int m = 0x5bd1e995; int r = 24; int h = seed ^ length; int len_4 = length >> 2; for (int i = 0; i < len_4; i++) { int i_4 = i << 2; int k = data[i_4 + 3]; k = k << 8; k = k | (data[i_4 + 2] & 0xff); k = k << 8; k = k | (data[i_4 + 1] & 0xff); k = k << 8; k = k | (data[i_4 + 0] & 0xff); k *= m; k ^= k >>> r; k *= m; h *= m; h ^= k; } // avoid calculating modulo int len_m = len_4 << 2; int left = length - len_m; if (left != 0) { if (left >= 3) { h ^= (int) data[length - 3] << 16; } if (left >= 2) { h ^= (int) data[length - 2] << 8; } if (left >= 1) { h ^= (int) data[length - 1]; } h *= m; } h ^= h >>> 13; h *= m; h ^= h >>> 15; return h; } public static int hashLong(long data) { int m = 0x5bd1e995; int r = 24; int h = 0; int k = (int) data * m; k ^= k >>> r; h ^= k * m; k = (int) (data >> 32) * m; k ^= k >>> r; h *= m; h ^= k * m; h ^= h >>> 13; h *= m; h ^= h >>> 15; return h; } public static long hash64(Object o) { if (o == null) { return 0l; } else if (o instanceof String) { final byte[] bytes = ((String) o).getBytes(); return hash64(bytes, bytes.length); } else if (o instanceof byte[]) { final byte[] bytes = (byte[]) o; return hash64(bytes, bytes.length); } return hash64(o.toString()); } // 64 bit implementation copied from here: https://github.com/tnm/murmurhash-java /** * Generates 64 bit hash from byte array with default seed value. * * @param data byte array to hash * @param length length of the array to hash * @return 64 bit hash of the given string */ public static long hash64(final byte[] data, int length) { return hash64(data, length, 0xe17a1465); } /** * Generates 64 bit hash from byte array of the given length and seed. * * @param data byte array to hash * @param length length of the array to hash * @param seed initial seed value * @return 64 bit hash of the given array */ public static long hash64(final byte[] data, int length, int seed) { final long m = 0xc6a4a7935bd1e995L; final int r = 47; long h = (seed & 0xffffffffl) ^ (length * m); int length8 = length / 8; for (int i = 0; i < length8; i++) { final int i8 = i * 8; long k = ((long) data[i8 + 0] & 0xff) + (((long) data[i8 + 1] & 0xff) << 8) + (((long) data[i8 + 2] & 0xff) << 16) + (((long) data[i8 + 3] & 0xff) << 24) + (((long) data[i8 + 4] & 0xff) << 32) + (((long) data[i8 + 5] & 0xff) << 40) + (((long) data[i8 + 6] & 0xff) << 48) + (((long) data[i8 + 7] & 0xff) << 56); k *= m; k ^= k >>> r; k *= m; h ^= k; h *= m; } switch (length % 8) { case 7: h ^= (long) (data[(length & ~7) + 6] & 0xff) << 48; case 6: h ^= (long) (data[(length & ~7) + 5] & 0xff) << 40; case 5: h ^= (long) (data[(length & ~7) + 4] & 0xff) << 32; case 4: h ^= (long) (data[(length & ~7) + 3] & 0xff) << 24; case 3: h ^= (long) (data[(length & ~7) + 2] & 0xff) << 16; case 2: h ^= (long) (data[(length & ~7) + 1] & 0xff) << 8; case 1: h ^= (long) (data[length & ~7] & 0xff); h *= m; } ; h ^= h >>> r; h *= m; h ^= h >>> r; return h; } }