HyperLogLog
HyperLogLog 源码分析(基数统计)
HyperLogLog 解决的问题:用固定的小内存估算海量数据的基数(不重复元素个数),典型场景是站点 UV。
核心思路:每个元素做 64 位 hash,低 14 位选桶,剩下 50 位数”末尾连续 0 的长度”,16384 个桶各自只记见过的最大长度,最后对全部桶做调和平均反推基数。
出现超长连零是小概率事件,观察到它说明来过很多不同元素;单桶方差大,一万六千个桶平均下来,12KB 换约 0.81% 的标准误差。
注意:2017 年起 hllCount 换成了 Ertl 论文(arXiv:1702.01284)的 tau/sigma 估计器,网上大量教程还在讲原论文那套”调和平均 + 小基数换线性计数 + 大基数 2^32 修正”,那些分段 if 在现在的源码里已经不存在了。
先看骨架:HLL 没有独立的对象类型,它就是一个带 16 字节头的普通 String。
代码块收起展开
// 基于本地 Redis 仓 (unstable, 8.4 之后), src/hyperloglog.c
struct hllhdr {
char magic[4]; /* "HYLL" */ // 就存在普通 String 里，靠魔数区分"这是个 HLL"
uint8_t encoding; /* HLL_DENSE or HLL_SPARSE. */
uint8_t notused[3]; /* Reserved for future use, must be zero. */
uint8_t card[8]; /* Cached cardinality, little endian. */ // 基数缓存：桶没变时 PFCOUNT 直接 O(1) 返回
uint8_t registers[]; /* Data bytes. */ // dense 下是 16384 个 6bit 寄存器紧密排列
};
/* The cached cardinality MSB is used to signal validity of the cached value. */
#define HLL_INVALIDATE_CACHE(hdr) (hdr)->card[7] |= (1<<7) // 借 64 位缓存的最高位当脏标记，基数不可能真用到 2^63
#define HLL_VALID_CACHE(hdr) (((hdr)->card[7] & (1<<7)) == 0)
#define HLL_P 14 /* The greater is P, the smaller the error. */ // 误差 = 1.04/sqrt(2^P)，P=14 是内存与精度的平衡点
#define HLL_Q (64-HLL_P) /* The number of bits of the hash value used for
determining the number of leading zeros. */
#define HLL_REGISTERS (1<<HLL_P) /* With P=14, 16384 registers. */
#define HLL_P_MASK (HLL_REGISTERS-1) /* Mask to index register. */
#define HLL_BITS 6 /* Enough to count up to 63 leading zeroes. */ // 桶值最大 Q+1=51，5 bit 装不下，6 bit 刚好
#define HLL_REGISTER_MAX ((1<<HLL_BITS)-1)
#define HLL_HDR_SIZE sizeof(struct hllhdr)
#define HLL_DENSE_SIZE (HLL_HDR_SIZE+((HLL_REGISTERS*HLL_BITS+7)/8)) // 16384*6/8 = 12288 字节，"12KB"的出处
#define HLL_DENSE 0 /* Dense encoding. */
#define HLL_SPARSE 1 /* Sparse encoding. */
#define HLL_RAW 255 /* Only used internally, never exposed. */ // 多 key PFCOUNT 临时用的一桶一字节格式
// ...
/* Note: if we access the last counter, we will also access the b+1 byte
* that is out of the array, but sds strings always have an implicit null
* term, so the byte exists, and we can skip the conditional (or the need
* to allocate 1 byte more explicitly). */
#define HLL_DENSE_SET_REGISTER(p,regnum,val) do { \
uint8_t *_p = (uint8_t*) p; \
unsigned long _byte = (regnum)*HLL_BITS/8; \
unsigned long _fb = (regnum)*HLL_BITS&7; \
unsigned long _fb8 = 8 - _fb; \
unsigned long _v = (val); \
_p[_byte] &= ~(HLL_REGISTER_MAX << _fb); \
_p[_byte] |= _v << _fb; \
_p[_byte+1] &= ~(HLL_REGISTER_MAX >> _fb8); \
_p[_byte+1] |= _v >> _fb8; \
} while(0)
// ...
int hllPatLen(unsigned char *ele, size_t elesize, long *regp) {
uint64_t hash, index;
int count;
// ...
hash = MurmurHash64A(ele,elesize,0xadc83b19ULL); // 种子写死：同一元素永远同一 hash，PFADD 才可能幂等
index = hash & HLL_P_MASK; /* Register index. */ // 低 14 位选桶
hash >>= HLL_P; /* Remove bits used to address the register. */ // 选桶用过的位必须丢掉，不能再参与数零
hash |= ((uint64_t)1<<HLL_Q); /* Make sure the loop terminates
and count will be <= Q+1. */ // 第 50 位人工补个 1，剩余位全 0 时 ctz 也有界
count = __builtin_ctzll(hash) + 1; // 数末尾连零，一条硬件指令；+1 是把结尾那个 1 也计入
*regp = (int) index;
return count;
}
int hllDenseSet(uint8_t *registers, long index, uint8_t count) {
uint8_t oldcount;
HLL_DENSE_GET_REGISTER(oldcount,registers,index);
if (count > oldcount) {
HLL_DENSE_SET_REGISTER(registers,index,count); // 只增不减地记最大值，幂等与可合并都来自这一行
return 1; // 1 = 桶变了，上层据此失效缓存、传播命令
} else {
return 0;
}
}
int hllDenseAdd(uint8_t *registers, unsigned char *ele, size_t elesize) {
long index;
uint8_t count = hllPatLen(ele,elesize,&index);
/* Update the register if this element produced a longer run of zeroes. */
return hllDenseSet(registers,index,count);
}
/* Call hllDenseAdd() or hllSparseAdd() according to the HLL encoding. */
int hllAdd(robj *o, unsigned char *ele, size_t elesize) {
struct hllhdr *hdr = o->ptr;
switch(hdr->encoding) {
case HLL_DENSE: return hllDenseAdd(hdr->registers,ele,elesize);
case HLL_SPARSE: return hllSparseAdd(o,ele,elesize); // sparse 版同样是 hllPatLen + hllSparseSet
default: return -1; /* Invalid representation. */
}
}小基数时 12KB 太浪费,稀疏编码用三种 opcode 做游程压缩:大段空桶压成一两个字节,只有非零桶才占地方。新建的 HLL 全部走 sparse。
代码块收起展开
// 基于本地 Redis 仓 (unstable, 8.4 之后), src/hyperloglog.c
#define HLL_SPARSE_XZERO_BIT 0x40 /* 01xxxxxx */
#define HLL_SPARSE_VAL_BIT 0x80 /* 1vvvvvxx */
#define HLL_SPARSE_IS_ZERO(p) (((*(p)) & 0xc0) == 0) /* 00xxxxxx */ // ZERO: 1 字节，表示连续 1~64 个空桶
#define HLL_SPARSE_IS_XZERO(p) (((*(p)) & 0xc0) == HLL_SPARSE_XZERO_BIT) // XZERO: 2 字节，表示连续 1~16384 个空桶
#define HLL_SPARSE_IS_VAL(p) ((*(p)) & HLL_SPARSE_VAL_BIT) // VAL: 1 字节 = 5 bit 桶值 + 2 bit 游程长度
// ...
#define HLL_SPARSE_VAL_MAX_VALUE 32 // 5 bit 只能表示 1~32，桶值一旦超过 32 稀疏编码就存不下了
#define HLL_SPARSE_VAL_MAX_LEN 4
#define HLL_SPARSE_ZERO_MAX_LEN 64
#define HLL_SPARSE_XZERO_MAX_LEN 16384
// ...
/* Create an HLL object. We always create the HLL using sparse encoding.
* This will be upgraded to the dense representation as needed. */
robj *createHLLObject(void) {
robj *o;
struct hllhdr *hdr;
sds s;
uint8_t *p;
int sparselen = HLL_HDR_SIZE +
(((HLL_REGISTERS+(HLL_SPARSE_XZERO_MAX_LEN-1)) /
HLL_SPARSE_XZERO_MAX_LEN)*2); // 16 字节头 + 1 条 XZERO = 18 字节
int aux;
/* Populate the sparse representation with as many XZERO opcodes as
* needed to represent all the registers. */
aux = HLL_REGISTERS;
s = sdsnewlen(NULL,sparselen);
p = (uint8_t*)s + HLL_HDR_SIZE;
while(aux) {
int xzero = HLL_SPARSE_XZERO_MAX_LEN;
if (xzero > aux) xzero = aux;
HLL_SPARSE_XZERO_SET(p,xzero); // 一条 XZERO 就能表示"16384 个桶全 0"
p += 2;
aux -= xzero;
}
serverAssert((p-(uint8_t*)s) == sparselen);
/* Create the actual object. */
o = createObject(OBJ_STRING,s); // 类型就是 OBJ_STRING，GET/SET 都能碰它，这是刻意设计
hdr = o->ptr;
memcpy(hdr->magic,"HYLL",4);
hdr->encoding = HLL_SPARSE; // 一律 sparse 起步：新 HLL 十几字节，12KB 是上限价不是起步价
return o;
}
int hllSparseSet(robj *o, long index, uint8_t count) {
struct hllhdr *hdr;
uint8_t oldcount, *sparse, *end, *p, *prev, *next;
long first, span;
long is_zero = 0, is_xzero = 0, is_val = 0, runlen = 0;
/* If the count is too big to be representable by the sparse representation
* switch to dense representation. */
if (count > HLL_SPARSE_VAL_MAX_VALUE) goto promote; // 升级触发点一：桶值超 32
// ... (先按需扩容 sds：最坏情况 XZERO 裂成 XZERO-VAL-XZERO 多 3 字节；
// 然后从头顺序扫 opcode，累加各游程覆盖的桶数，定位 index 落在哪个 opcode 里)
if (is_val) {
oldcount = HLL_SPARSE_VAL_VALUE(p);
/* Case A. */
if (oldcount >= count) return 0; // 最高频路径：老值已经够大，直接返回 0
/* Case B. */
if (runlen == 1) {
HLL_SPARSE_VAL_SET(p,count,1); // 该 VAL 恰好只盖这一个桶，原地改值
goto updated;
}
}
/* C) Another trivial to handle case is a ZERO opcode with a len of 1.
* We can just replace it with a VAL opcode with our value and len of 1. */
if (is_zero && runlen == 1) {
HLL_SPARSE_VAL_SET(p,count,1);
goto updated;
}
// ... (D 一般情况：把命中的游程切成最多三段(如 XZERO-VAL-XZERO，最长 5 字节)写入栈上 seq，
// 比旧 opcode 长出 deltalen 字节就 memmove 右侧数据腾位，再把 seq 拷回去)
int seqlen = n-seq;
int oldlen = is_xzero ? 2 : 1;
int deltalen = seqlen-oldlen;
if (deltalen > 0 &&
sdslen(o->ptr) + deltalen > server.hll_sparse_max_bytes) goto promote; // 升级触发点二：超过 hll-sparse-max-bytes(默认 3000)
// ...
updated:
// ... (Step 4: 从 prev 起最多扫 5 个 opcode，相邻同值 VAL 能合并就合并，回收字节)
/* Invalidate the cached cardinality. */
hdr = o->ptr;
HLL_INVALIDATE_CACHE(hdr); // 桶变了，PFCOUNT 缓存作废
return 1;
promote: /* Promote to dense representation. */
if (hllSparseToDense(o) == C_ERR) return -1; /* Corrupted HLL. */ // 单向升级，dense 永不退回 sparse
hdr = o->ptr;
// ...
int dense_retval = hllDenseSet(hdr->registers,index,count); // 转完在 dense 上补做这次写入
serverAssert(dense_retval == 1);
return dense_retval;
}最后是读路径:估算基数与多 HLL 合并。新版给 dense 的合并/压缩还加了 AVX2/NEON 的 SIMD 实现(hllMergeDenseAVX2 等),下面看标量主干。
代码块收起展开
// 基于本地 Redis 仓 (unstable, 8.4 之后), src/hyperloglog.c
uint64_t hllCount(struct hllhdr *hdr, int *invalid) {
double m = HLL_REGISTERS;
double E;
int j;
// ...
int reghisto[64] = {0}; // 不关心每个桶的具体值，只要"值为 k 的桶有几个"这张直方图
/* Compute register histogram */
if (hdr->encoding == HLL_DENSE) {
hllDenseRegHisto(hdr->registers,reghisto);
} else if (hdr->encoding == HLL_SPARSE) {
hllSparseRegHisto(hdr->registers,
sdslen((sds)hdr)-HLL_HDR_SIZE,invalid,reghisto); // sparse 不解码成桶数组，扫 opcode 直接出直方图
} else if (hdr->encoding == HLL_RAW) {
hllRawRegHisto(hdr->registers,reghisto);
} else {
serverPanic("Unknown HyperLogLog encoding in hllCount()");
}
/* Estimate cardinality from register histogram. See:
* "New cardinality estimation algorithms for HyperLogLog sketches"
* Otmar Ertl, arXiv:1702.01284 */
double z = m * hllTau((m-reghisto[HLL_Q+1])/(double)m); // tau 项修正饱和桶(超大基数端)
for (j = HLL_Q; j >= 1; --j) {
z += reghisto[j];
z *= 0.5; // 从高到低累加折半，等价于 sum(count_k * 2^-k)，即调和平均的分母
}
z += m * hllSigma(reghisto[0]/(double)m); // sigma 项修正空桶(小基数端)，全程无分段 if
E = llroundl(HLL_ALPHA_INF*m*m/z); // alpha * m^2 / z，alpha 约 0.7213 用于消系统性偏差
return (uint64_t) E;
}
void hllMergeDense(uint8_t* reg_raw, const uint8_t* reg_dense) {
// ... (默认配置且 CPU 支持时走 hllMergeDenseAVX2 / hllMergeDenseAarch64)
uint8_t val;
for (int i = 0; i < HLL_REGISTERS; i++) {
HLL_DENSE_GET_REGISTER(val, reg_dense, i);
reg_raw[i] = MAX(reg_raw[i], val); // 合并 = 逐桶取 max，就这一行
}
}原理串讲
一次 PFADD key a b c 走完整链路。入口 pfaddCommand:key 不存在就 createHLLObject,新对象是 sparse 编码,16 字节头加一条 XZERO 共 18 字节;然后对每个元素调 hllAdd,按 hdr->encoding 分发到 hllSparseAdd 或 hllDenseAdd,两条路都是先 hllPatLen 算观测值,再 set 进桶。
hllPatLen 是算法核心。MurmurHash64A 的种子写死为 0xadc83b19,同一元素永远得到同一 hash,于是重复添加会算出完全相同的 (index, count),而桶里记的是 max,重复写入等于没写。
第一处关键设计就在这:HLL 的”去重”不需要存任何元素,幂等性是 hash 确定性加 max 语义自动推出来的。
取完低 14 位桶号后立刻 hash >>= HLL_P 把这些位丢掉,选桶的位不能再参与数零,否则桶号和桶值相关,16384 个”独立实验”就不独立了,误差分析全部失效。
接着 hash |= 1<<HLL_Q 在第 50 位人工补一个 1,保证 __builtin_ctzll 有界(count 最大 Q+1=51),用一次或运算换掉循环和特判。
sparse 路径进 hllSparseSet:顺序扫 opcode 定位目标桶,命中 VAL 且老值更大就直接返回 0(最高频路径放最前);只有游程为 1 的 VAL/ZERO 能原地改,其余都要把游程切成最多三段再 memmove 插回。
两个 goto promote 是仅有的升级入口:桶值超 32(VAL 的 5 bit 存不下),或稀疏串要超过 hll-sparse-max-bytes(默认 3000 字节)。
为什么单向升级、永不回退?桶值只增不减,稀疏度只会越来越差,回退没有收益,而 hllSparseToDense 的一次性转换成本摊在整个生命周期里可以忽略。
升级后所有读写走 HLL_DENSE_SET_REGISTER 这类位运算宏:6 bit 桶不按字节对齐,一个桶可能横跨两个字节,宏无条件读写 _byte 和 _byte+1,最后一个桶会越界碰到数组外 1 字节,靠 sds 隐式的结尾 ‘\0’ 保证那个字节合法,用一个免费字节换掉热路径上的边界分支。
PFCOUNT key 先查 HLL_VALID_CACHE:任何一次成功的桶更新都会 HLL_INVALIDATE_CACHE 置脏 card[7] 最高位,缓存有效时把 card 的 8 字节小端拼出来直接返回,O(1)。
脏了才调 hllCount 重算并写回缓存。缓存放在对象内部而非某个内存字段,是因为 HLL 就是个 String 值,缓存跟着值走,RDB 落盘、主从复制天然带上;代价是 PFCOUNT 这个”读命令”可能修改值,所以源码里专门 keyModified 加 server.dirty++ 把缓存更新传播出去。
多 key 的 PFCOUNT k1 k2 在栈上开 16KB 的 HLL_RAW 表示(一桶一字节),把每个 HLL 用 hllMerge 逐桶取 max 合进去再 hllCount,避免中间结果反复解 6 bit。
PFMERGE 同理。为什么 max 合并是无损的?两个集合并集的”每桶最大连零观测”恰好等于各自观测的 max,数学上严格相等,所以分布式场景可以各机器独立统计、汇总时 merge,结果和单机统计完全一致。
估算本身在 hllCount:先出直方图,再从高位到低位 z += reghisto[j]; z *= 0.5,等价于 sum(count_k * 2^-k),也就是调和平均的分母。
第二处关键设计:用调和平均而非算术平均,因为它对离群大值钝感,个别桶偶然撞上超长连零(观测值意味着 2^50 级别)不会把整体估算拉飞。
sigma 和 tau 两个级数分别把空桶(小基数端)和饱和桶(超大基数端)的偏差修正吸收进同一个公式,替代了旧版按估算值大小切换算法的分段逻辑。
设计取舍
- 每桶记 max 而非计数:换来幂等与无损合并,代价是只能估基数,取不出元素、也判不了”某元素是否存在”(那是布隆过滤器的活)。
- sparse 起步、单向升级:基数几千以内 sparse 只占几百字节;调大 hll-sparse-max-bytes 更省内存,但 sparse 更新是 memmove 密集的 O(n) 扫描,大了反而拖慢 PFADD。
- 基数缓存借 card[7] 最高位当脏位,不多花一个字节;副作用是 PFCOUNT 读命令也会产生复制流量,属于 Redis 里少见的”读命令改值”。
- HLL 是 OBJ_STRING:能 GET 出来备份、SET 到别处恢复,AOF/RDB 零改动;代价是用户可能 SET 进坏数据,所以每个命令入口都过
isHLLObjectOrReply魔数校验,损坏时报 INVALIDOBJ。 - 误差 1.04/sqrt(m):P 每加 1 内存翻倍,误差只降 sqrt(2) 倍,收益递减,P=14 的 12KB/0.81% 是工程平衡点。