走近源码：神奇的HyperLogLog

  • 2020 年 3 月 11 日
  • 笔记

HyperLogLog是Redis的高级数据结构，是统计基数的利器。前文我们已经介绍过HyperLogLog的基本用法，如果只求会用，只需要掌握HyperLogLog的三个命令即可，如果想要更进一步了解HyperLogLog的原理以及源码实现，相信这篇文章会给你带来一些启发。

基数

在数学上，基数或势，即集合中包含的元素的“个数”（参见势的比较），是日常交流中基数的概念在数学上的精确化（并使之不再受限于有限情形）。有限集合的基数，其意义与日常用语中的“基数”相同，例如{displaystyle {a,b,c}}的基数是3。无限集合的基数，其意义在于比较两个集的大小，例如整数集和有理数集的基数相同；整数集的基数比实数集的小。

在介绍HyperLogLog的原理之前，请你先来思考一下，如果让你来统计基数，你会用什么方法。

Set

熟悉Redis数据结构的同学一定首先会想到Set这个结构，我们只需要把数据都存入Set，然后用scard命令就可以得到结果，这是一种思路，但是存在一定的问题。如果数据量非常大，那么将会耗费很大的内存空间，如果这些数据仅仅是用来统计基数，那么无疑是造成了巨大的浪费，因此，我们需要找到一种占用内存较小的方法。

bitmap

bitmap同样是一种可以统计基数的方法，可以理解为用bit数组存储元素，例如01101001，表示的是[1,2,4,8]，bitmap中1的个数就是基数。bitmap也可以轻松合并多个集合，只需要将多个数组进行异或操作就可以了。bitmap相比于Set也大大节省了内存，我们来粗略计算一下，统计1亿个数据的基数，需要的内存是：100000000/8/1024/1024 ≈ 12M。

虽然bitmap在节省空间方面已经有了不错的表现，但是如果需要统计1000个对象，就需要大约12G的内存，显然这个结果仍然不能令我们满意。在这种情况下，HyperLogLog将会出来拯救我们。

HyperLogLog原理

HyperLogLog实际上不会存储每个元素的值，它使用的是概率算法，通过存储元素的hash值的第一个1的位置，来计算元素数量。这么说不太容易理解，容我先搬出来一个栗子。

有一天Jack和丫丫玩抛硬币的游戏，规则是丫丫负责抛硬币，每次抛到正面为一回合，丫丫可以自己决定进行几个回合。最后需要告诉Jack最长的那个回合抛了多少次，再由Jack来猜丫丫一共进行了几个回合。Jack心想：这可不好猜啊，我得算算概率了。于是在脑海中绘制这样一张图。

yb

k是每回合抛到1所用的次数，我们已知的是最大的k值，可以用kmax表示，由于每次抛硬币的结果只有0和1两种情况，因此，kmax在任意回合出现的概率即为(1/2)kmax，因此可以推测n=2kmax。概率学把这种问题叫做伯努利实验。此时丫丫已经完成了n个回合，并且告诉Jack最长的一次抛了3次，Jack此时也胸有成竹，马上说出他的答案8，最后的结果是：丫丫只抛了一回合，Jack输了，要负责刷碗一个月。

终于，我们的Philippe Flajolet教授遇到了Jack一样的问题，他决心吸取Jack的教训，要让这个算法更加准确，于是引入了桶的概念，计算m个桶的加权平均值，这样就能得到比较准确的答案了（实际上还要进行其他修正）。最终的公式如图

HyperLogLog公式

其中m是桶的数量，const是修正常数，它的取值会根据m而变化。p=log2m

switch (p) {     case 4:         constant = 0.673 * m * m;     case 5:         constant = 0.697 * m * m;     case 6:         constant = 0.709 * m * m;     default:         constant = (0.7213 / (1 + 1.079 / m)) * m * m;  }

我们回到Redis，对于一个输入的字符串，首先得到64位的hash值，用前14位来定位桶的位置（共有214，即16384个桶）。后面50位即为伯努利过程，每个桶有6bit，记录第一次出现1的位置count，如果count>oldcount，就用count替换oldcount。

了解原理之后，我们再来聊一下HyperLogLog的存储。HyperLogLog的存储结构分为密集存储结构和稀疏存储结构两种，默认为稀疏存储结构，而我们常说的占用12K内存的则是密集存储结构。

密集存储结构

密集存储比较简单，就是连续16384个6bit的串成的位图。由于每个桶是6bit，因此对桶的定位要麻烦一些。

#define HLL_BITS 6 /* Enough to count up to 63 leading zeroes. */  #define HLL_REGISTER_MAX ((1<<HLL_BITS)-1)  /* Store the value of the register at position 'regnum' into variable 'target'.   * 'p' is an array of unsigned bytes. */  #define HLL_DENSE_GET_REGISTER(target,p,regnum) 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 b0 = _p[_byte];       unsigned long b1 = _p[_byte+1];       target = ((b0 >> _fb) | (b1 << _fb8)) & HLL_REGISTER_MAX;   } while(0)    /* Set the value of the register at position 'regnum' to 'val'.   * 'p' is an array of unsigned bytes. */  #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)

如果我们要定位的桶编号为regnum，它的偏移字节量为(regnum * 6) / 8，起始bit偏移为(regnum * 6) % 8，例如，我们要定位编号为5的桶，字节偏移是3，位偏移也是6，也就是说，从第4个字节的第7位开始是编号为3的桶。这里需要注意，字节序和我们平时的字节序相反，因此需要进行倒置。我们用一张图来说明Redis是如何定位桶并且得到存储的值（即HLL_DENSE_GET_REGISTER函数的解释）。

桶定位

对于编号为5的桶，我们已经得到了字节偏移_byte和为偏移_fb，b0 >> _fb和b1 << _fb8操作是将字节倒置，然后进行拼接，并且保留最后6位。

稀疏存储结构

你以为Redis真的会用16384个6bit存储每一个HLL对象吗，那就too naive了，虽然它只占用了12K内存，但是Redis对于内存的节约已经到了丧心病狂的地步了。因此，如果比较多的计数值都是0，那么就会采用稀疏存储的结构。

对于连续多个计数值为0的桶，Redis使用的存储方式是：00xxxxxx，前缀两个0，后面6位的值加1表示有连续多少个桶的计数值为0，由于6bit最大能表示64个桶，所以Redis又设计了另一种表示方法：01xxxxxx yyyyyyyy，这样后面14bit就可以表示16384个桶了，而一个初始状态的HyperLogLog对象只需要用2个字节来存储。

如果连续的桶数都不是0，那么Redis的表示方式为1vvvvvxx，即为连续(xx+1)个桶的计数值都是(vvvvv+1)。例如，10011110表示连续3个8。这里使用5bit，最大只能表示32。因此，当某个计数值大于32时，Redis会将这个HyperLogLog对象调整为密集存储。

Redis用三条指令来表达稀疏存储的方式：

  1. ZERO:len 单个字节表示 00[len-1]，连续最多64个零计数值
  2. VAL:value,len 单个字节表示 1[value-1][len-1]，连续 len 个值为 value 的计数值
  3. XZERO:len 双字节表示 01[len-1]，连续最多16384个零计数值

Redis从稀疏存储转换到密集存储的条件是：

  1. 任意一个计数值从 32 变成 33，因为VAL指令已经无法容纳，它能表示的计数值最大为 32
  2. 稀疏存储占用的总字节数超过 3000 字节，这个阈值可以通过 hll_sparse_max_bytes 参数进行调整。

源码解析

接下来通过源码来看一下pfadd和pfcount两个命令的具体流程。在这之前我们首先要了解的是HyperLogLog的头结构体和创建一个HyperLogLog对象的步骤。

HyperLogLog头结构体
struct hllhdr {      char magic[4];      /* "HYLL" */      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. */      uint8_t registers[]; /* Data bytes. */  };
创建HyperLogLog对象
#define HLL_P 14 /* The greater is P, the smaller the error. */  #define HLL_REGISTERS (1<<HLL_P) /* With P=14, 16384 registers. */  #define HLL_SPARSE_XZERO_MAX_LEN 16384      #define HLL_SPARSE_XZERO_SET(p,len) do {       int _l = (len)-1;       *(p) = (_l>>8) | HLL_SPARSE_XZERO_BIT;       *((p)+1) = (_l&0xff);   } while(0)    /* 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);      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);          p += 2;          aux -= xzero;      }      serverAssert((p-(uint8_t*)s) == sparselen);        /* Create the actual object. */      o = createObject(OBJ_STRING,s);      hdr = o->ptr;      memcpy(hdr->magic,"HYLL",4);      hdr->encoding = HLL_SPARSE;      return o;  }

这里sparselen=HLL_HDR_SIZE+2，因为初始化时默认所有桶的计数值都是0。其他过程不难理解，用的存储方式是我们前面提到过的稀疏存储，创建的对象实质上是一个字符串对象，这也是字符串命令可以操作HyperLogLog对象的原因。

PFADD命令
/* PFADD var ele ele ele ... ele => :0 or :1 */  void pfaddCommand(client *c) {      robj *o = lookupKeyWrite(c->db,c->argv[1]);      struct hllhdr *hdr;      int updated = 0, j;        if (o == NULL) {          /* Create the key with a string value of the exact length to           * hold our HLL data structure. sdsnewlen() when NULL is passed           * is guaranteed to return bytes initialized to zero. */          o = createHLLObject();          dbAdd(c->db,c->argv[1],o);          updated++;      } else {          if (isHLLObjectOrReply(c,o) != C_OK) return;          o = dbUnshareStringValue(c->db,c->argv[1],o);      }      /* Perform the low level ADD operation for every element. */      for (j = 2; j < c->argc; j++) {          int retval = hllAdd(o, (unsigned char*)c->argv[j]->ptr,                                 sdslen(c->argv[j]->ptr));          switch(retval) {          case 1:              updated++;              break;          case -1:              addReplySds(c,sdsnew(invalid_hll_err));              return;          }      }      hdr = o->ptr;      if (updated) {          signalModifiedKey(c->db,c->argv[1]);          notifyKeyspaceEvent(NOTIFY_STRING,"pfadd",c->argv[1],c->db->id);          server.dirty++;          HLL_INVALIDATE_CACHE(hdr);      }      addReply(c, updated ? shared.cone : shared.czero);  }

PFADD命令会先判断key是否存在，如果不存在，则创建一个新的HyperLogLog对象；如果存在，会调用isHLLObjectOrReply()函数检查这个对象是不是HyperLogLog对象，检查方法主要是检查魔数是否正确，存储结构是否正确以及头结构体的长度是否正确等。

一切就绪后，才可以调用hllAdd()函数添加元素。hllAdd函数很简单，只是根据存储结构判断需要调用hllDenseAdd()函数还是hllSparseAdd()函数。

密集存储结构只是比较新旧计数值，如果新计数值大于就计数值，就将其替代。

而稀疏存储结构要复杂一些：

  1. 判断是否需要调整为密集存储结构，如果不需要则继续进行，否则就先调整为密集存储结构，然后执行添加操作
  2. 我们需要先定位要修改的字节段，通过循环计算每一段表示的桶的范围是否包括要修改的桶
  3. 定位到桶后，如果这个桶已经是VAL，并且计数值大于当前要添加的计数值，则返回0，如果小于当前计数值，就进行更新
  4. 如果是ZERO，并且长度为1，那么可以直接把它替换为VAL，并且设置计数值
  5. 如果不是上述两种情况，则需要对现有的存储进行拆分
PFCOUNT命令
/* PFCOUNT var -> approximated cardinality of set. */  void pfcountCommand(client *c) {      robj *o;      struct hllhdr *hdr;      uint64_t card;        /* Case 1: multi-key keys, cardinality of the union.       *       * When multiple keys are specified, PFCOUNT actually computes       * the cardinality of the merge of the N HLLs specified. */      if (c->argc > 2) {          uint8_t max[HLL_HDR_SIZE+HLL_REGISTERS], *registers;          int j;            /* Compute an HLL with M[i] = MAX(M[i]_j). */          memset(max,0,sizeof(max));          hdr = (struct hllhdr*) max;          hdr->encoding = HLL_RAW; /* Special internal-only encoding. */          registers = max + HLL_HDR_SIZE;          for (j = 1; j < c->argc; j++) {              /* Check type and size. */              robj *o = lookupKeyRead(c->db,c->argv[j]);              if (o == NULL) continue; /* Assume empty HLL for non existing var.*/              if (isHLLObjectOrReply(c,o) != C_OK) return;                /* Merge with this HLL with our 'max' HHL by setting max[i]               * to MAX(max[i],hll[i]). */              if (hllMerge(registers,o) == C_ERR) {                  addReplySds(c,sdsnew(invalid_hll_err));                  return;              }          }            /* Compute cardinality of the resulting set. */          addReplyLongLong(c,hllCount(hdr,NULL));          return;      }        /* Case 2: cardinality of the single HLL.       *       * The user specified a single key. Either return the cached value       * or compute one and update the cache. */      o = lookupKeyWrite(c->db,c->argv[1]);      if (o == NULL) {          /* No key? Cardinality is zero since no element was added, otherwise           * we would have a key as HLLADD creates it as a side effect. */          addReply(c,shared.czero);      } else {          if (isHLLObjectOrReply(c,o) != C_OK) return;          o = dbUnshareStringValue(c->db,c->argv[1],o);            /* Check if the cached cardinality is valid. */          hdr = o->ptr;          if (HLL_VALID_CACHE(hdr)) {              /* Just return the cached value. */              card = (uint64_t)hdr->card[0];              card |= (uint64_t)hdr->card[1] << 8;              card |= (uint64_t)hdr->card[2] << 16;              card |= (uint64_t)hdr->card[3] << 24;              card |= (uint64_t)hdr->card[4] << 32;              card |= (uint64_t)hdr->card[5] << 40;              card |= (uint64_t)hdr->card[6] << 48;              card |= (uint64_t)hdr->card[7] << 56;          } else {              int invalid = 0;              /* Recompute it and update the cached value. */              card = hllCount(hdr,&invalid);              if (invalid) {                  addReplySds(c,sdsnew(invalid_hll_err));                  return;              }              hdr->card[0] = card & 0xff;              hdr->card[1] = (card >> 8) & 0xff;              hdr->card[2] = (card >> 16) & 0xff;              hdr->card[3] = (card >> 24) & 0xff;              hdr->card[4] = (card >> 32) & 0xff;              hdr->card[5] = (card >> 40) & 0xff;              hdr->card[6] = (card >> 48) & 0xff;              hdr->card[7] = (card >> 56) & 0xff;              /* This is not considered a read-only command even if the               * data structure is not modified, since the cached value               * may be modified and given that the HLL is a Redis string               * we need to propagate the change. */              signalModifiedKey(c->db,c->argv[1]);              server.dirty++;          }          addReplyLongLong(c,card);      }  }

如果要计算多个HyperLogLog的基数，则需要将多个HyperLogLog对象合并，这里合并方法是将所有的HyperLogLog对象合并到一个名为max的对象中，max采用的是密集存储结构，如果被合并的对象也是密集存储结构，则循环比较每一个计数值，将大的那个存入max。如果被合并的是稀疏存储，则只需要比较VAL即可。

如果计算单个HyperLogLog对象的基数，则先判断对象头结构体中的基数缓存是否有效，如果有效，可直接返回。如果已经失效，则需要重新计算基数，并修改原有缓存，这也是PFCOUNT命令不被当做只读命令的原因。

结语

最后，给大家推荐一个帮助理解HyperLogLog原理的工具：http://content.research.neustar.biz/blog/hll.html，有兴趣的话可以去学习一下。

英文比较好的同学可以直接点击阅读原文阅读antirez的关于HyperLogLogd 博客。

VirMach 便宜 VPS

QNews

QNews