Bitmap 续篇 - 基于 Bitmap 瞅瞅不一样的 Percentile

作者：杨承波

GrowingIO 大数据工程师，主要负责 SaaS 分析和广告模块的技术设计与开发，目前专注于 GrowingIO 物化视图引擎的建设。

本篇文章主要讲解基于 GrowingIO 内部数据存储结构 Bitmap 实现的 Percentile，并简单介绍一下 Hive Percentile，Spark Percentile，Bitmap Percentile 之间的差异。

讲解具体函数算法实现之前，先搞清楚下面两个问题：

• Percentile 函数【以下简称：分位数】，在实际应用场景下的作用是什么？

• 今天你是否还想起，Bitmap 是什么东西？

1. 分位数是个什么玩意儿🔢

1.1 分位数

😱分位数是啥？

针对一个从小到大的有序集合，找一个数来将集合按分位对应比例拆解成两个集合。

注意：分位数并非是指数据集合中的某个元素，而是找到一个数来拆解集合

🤔来点儿晕头转向的东西，90 分位怎么出？

这个数将样本集合分成左右两个集合，而且左边部分的集合占 90%，右边部分占 10%。

①元素个数为奇数的集合如何求 90 分位数？

原始集合：【35, 40, 41, 44, 45, 46, 49, 50, 53, 55, 58】

拆解集合：【35, 40, 41, 44, 45, 46, 49, 50, 53】, 55, 【58】

左边集合元素个数：9

右边集合元素个数：1

正好在原始集合中有这个数能将样本按 90 分位比例拆解成两个集合

所以上面样本的 90 分位是 55

②如果换成元素个数为偶数的集合呢？

原始集合：【40, 41, 44, 45, 46, 49, 50, 53, 55, 58】

拆解集合：【40, 41, 44, 45, 46, 49, 50, 53, 55】, X, 【58】

这个数存在于 【55 → 58】

X =  58 - (58 - 55) * 0.9 = 55.3

或 X =  55 + (58 - 55) * (1 - 0.9) = 55.3

90 分位含义：有 90%的数据小于等于 55.3，有 10% 的数据大于等于 55.3

1.2 在业务场景中的应用

👉现有【订单支付成功】事件，获取过去七天每个用户做过该事件总次数的 90 分位值，如下图：

👉还是【订单支付成功】事件，获取昨天每个用户在该事件下实际购买金额的总和，求 75 分位值，如下图：

2. Percentile 哪家强？

2.1 性能对比

环境准备：Core[16], 内存 2G

对比测试：[基于 Bitmap 实现的 Percentile] VS [SparkSQL 内置 Percentile_approx]

场景：一定数量的用户以及随机生成对应的 count，随机生成分位进行计算分位数，获取百次平均消耗

x 轴含义: 数据量

y 轴含义: 计算时间, 单位毫秒。

1. SparkSQL 中 Percentile 仅支持 Int, Long 数据类型，这里为通用考虑，使用 Percentile_approx 进行对比

2. 以上 Bitmap 存储数据与 Percentile_approx 处理数据完全一致

2.2 Hive Percentile

Hive Sql 中 Percentile 求解时针对的是一列进行操作，即表里的某一个字段，面对动不动几千万的数据处理，如果把每条数据全都加载到内存中，结局只有一个——卡死。

所以 Hive 需要在 UDAF 的计算中将数据进行压缩或预处理，那么 Mapper 是需要在生成时不断通过聚合计算更新，其内部实现基于 histogram。

Hive 的 percentile_approx 实现灵感出自《A Streaming Parallel Decision Tree Algorithm》，这篇论文提出 On-line Histogram Building 算法。

什么是 histogram？定义如下：

A histogram is a set of B pairs (called bins) of real numbers {(p1,m1),...,(pB,mB)}, where B is a preset constant integer.

在 histogram 的定义里面，有一个用于标识 bins 数量的常量 B。为何一定要引入这个常量？假设我们有一个简单的 sample 数据集（元素可重复）：

[1, 1, 1, 2, 2, 2, 3, 4, 4, 5, 6, 7, 8, 9, 9, 10, 10]

其 histogram 为：[(1, 3), (2, 3), (3, 1), (4, 2), (5, 1), (6, 1), (7, 1), (8, 1), (9, 2), (10, 2)]。

可以看出，这个 histogram 内部的 bins（数据点和频率一起构成的 pair） 数组长度实质上就是 sample 数据集的基数（不同元素的个数）。

histogram 的存储开销会随着 sample 数据集的基数线性增长，这意味着如果不做额外的优化，histogram 将无法适应超大规模数据集合的统计需求。常量 B 就是在这种背景下引入的，其目的在于控制 histogram 的 bins 数组长度（内存开销）。

Hive 的 Percentile_approx 由 GenericUDAFPercentileApprox 实现，其核心实现是在 histogram 的 bins 数组前面加上一个用于标识分位点的序列。其 merge 操作结果仅保留 histogram 序列，最后从 histogram 计算结果，源码如下：

/**   * Gets an approximate quantile value from the current histogram. Some popular   * quantiles are 0.5 (median), 0.95, and 0.98.   *   * @param q The requested quantile, must be strictly within the range (0,1).   * @return The quantile value.   */  public double quantile(double q) {    assert(bins != null && nusedbins > 0 && nbins > 0);    double sum = 0, csum = 0;    int b;    for(b = 0; b < nusedbins; b++)  {      sum += bins.get(b).y;    }    for(b = 0; b < nusedbins; b++) {      csum += bins.get(b).y;      if(csum / sum >= q) {        if(b == 0) {          return bins.get(b).x;        }
        csum -= bins.get(b).y;        double r = bins.get(b-1).x +          (q*sum - csum) * (bins.get(b).x - bins.get(b-1).x)/(bins.get(b).y);        return r;      }    }    return -1; // for Xlint, code will never reach here  }

2.3 Spark Percentile

Percentile

• 仅接受 Int, Long，精确计算，底层用 OpenHashMap 计数，然后排序 key。

OpenHashMap 为了加快速度，增加了一个假设：

• 所有数据只插入 Key /更新 Key，不删除 Key。

• 这个假设在大数据处理/统计的场景下，大多都是成立的。

• 可以去掉拉链表，使用线性探测的开放定址法来实现哈希表。

OpenHashMap 底层数据为 OpenHashSet，所以本质上是看 OpenHashSet 为啥快。

OpenHashSet 中用 BitSet (位图)来存储是否存在于集合中(位运算)，另一个数组存储实际数据，结构如下：

protected var _bitset = new BitSet(_capacity)protected var _data: Array[T] = _  _data = new Array[T](_capacity)

• 俩成员始终保持等长，_bitset 的下标 x 位置为 1 时，则 _data 的下标 x 位置中就有实际数据。

• 插入数据时，hash(key) 生成 pos，看 _bitset 中对应 pos 是否被占用，有则 ++pos。

OpenHashSet 快的原因:

1. 内存利用率高: 去掉了 8B 的指针结构，能够创建更大的哈希表，冲突减少。

2. 内存紧凑: 位图操作快。

Percentile_approx

接受 Int, Long, Double，近似计算。使用 GK 算法。

论文参见《Space-efficient Online Computation of Quantile Summaries》

底层实现通过 QuantileSummaries 实现，主要有两个成员变量：

sample: Array[Stat] : 存放桶，超过1000个桶的时候就压缩（生成新的三元组）；headSampled: ArrayBuffer[Double]：缓冲区，每次达到5000个，就排序后更新到sample.

主要思想是减少空间占用，spark 实现 merge sample 时甚至未处理 samples 已经有序，直接 sortBy：

// TODO: could replace full sort by ordered merge, the two lists are known to be sorted already. val res = (sampled ++ other.sampled).sortBy(_.value)  val comp = compressImmut(res, mergeThreshold = 2 * relativeError * count)  new QuantileSummaries(other.compressThreshold, other.relativeError, comp, other.count + count)

Stat 的定义:

/**  * Statistics from the Greenwald-Khanna paper.  * @param value the sampled value  * @param g the minimum rank jump from the previous value's minimum rank  * @param delta the maximum span of the rank.  */ case class Stats(value: Double, g: Int, delta: Int)

插入函数：每 N 个数，排序至少 1 次(merge 还有 1 次)，时间复杂度 O(NlogN)：

def insert(x: Double): QuantileSummaries = {    headSampled += x    if (headSampled.size >= defaultHeadSize) {      val result = this.withHeadBufferInserted      if (result.sampled.length >= compressThreshold) {        result.compress()      } else {        result      }    } else {      this    }  }

获取结果: 时间复杂度 O(n)

// Target rank    val rank = math.ceil(quantile * count).toInt    val targetError = math.ceil(relativeError * count)    // Minimum rank at current sample    var minRank = 0    var i = 1    while (i < sampled.length - 1) {      val curSample = sampled(i)      minRank += curSample.g      val maxRank = minRank + curSample.delta      if (maxRank - targetError <= rank && rank <= minRank + targetError) {        return Some(curSample.value)      }      i += 1    }

2.4 BitMap 小插曲

干饭人，你是否还记得凯哥之前的分享《GrowingIO 基于 BitMap 的海量数据分析 》

这里只是简单回顾 CBitmap 和补充一丢丢权重相关的东西

非数值型指标 Bitmap 怎么存储？

这里只用单个事件 + 单个维度分析，多维度的可参考之前的分享文档，中华传统功夫以点到为止。

生成 CBitmap 统计结果。

CBitmap: Map(short → Map(dimsSortId →  RoaringBitmap(uid)))

数值型指标 Bitmap 怎么存储？

上面我们讲述的都是统计的某个指标发生的次数比较小的整数类型数据，那么问题来了：

①当我统计的结果不是个整数呢，换句话说，我们统计的指标是订单金额呢？

②当我统计的结果都是整数，但是吧，有些东西喜欢按”K”作为单位，用你的小脑袋瓜想一想，统计结果中次数从 0 → 1024【bit 位 0 → 10】就存了个寂寞

计算精度，期望对所有值统计出一个公约数，从而减少存储量，这里引出权重 Weight 的概念：

例如  对于 100，200，300, 可以提出一个 100 作为公约数，只保存 1 2 3，  同理 0.01 0.02 0.03 也可以提出 0.01
以下部分不再详细讲解，有兴趣可以瞅瞅/**  * 但是我们实现无法知道数据的分布，只能预估一个值，具体预估流程：  *   1. 计算一个 high 和 low， high 可以认为是求 log10 + 1，也就是和 1 差的数量级，  *      low 可以认为是保证精度在 1e-4（精度可以修改） 以内至少要保留的位数，对于整数来说 low = 0  *   2. 根据所有的 high 和 low，统计一个相对合理的 high 和 low，只要这个 high 对应的数据占比高于平均占比的 1/10 即可  *   3. high 代表了最大需要将数据放大多少个数量级，low 代表最小可以将数据缩小多少个数量级，  *      求一个折中值 Math.max(high-n, -1*low) 这里的 n 会影响整数的精度，一开始是 6，后来改为了 7  *  * 例如有一系列数字：  *     10,000 100 10 0.1 0.01  * 可以求得对应的 high 和 low：  *     5       3   2  -1  -2  *     0       0   0   1   2  * 然后求得合并的 high = 5, low = 2  * 最后得到 weight = 0.1(n=6)， 0.01(n=7)  *  * 再举一个极端的例子：  *     10000...(20个0) 0.00000...1(20个0)  * 求得的 high 和 low:  *     21           -21  *     0             0  * 求得合并的 high = 21, low = 0  * 合并得到 weight= 100...(21-7个0)  */

2.5 Bitmap 下实现 Percentile

计算逻辑

1. 将 n 个变量从小到大排序为数组 x， p 是分位，x[i] 表示 x 数组的第 i 个元素

2. 设(n - 1) * p% = integerPart + decimalPart，integerPart 为整数部分，decimalPart 为小数部分，这里的 integerPart 其实是 x 的数组下标，是从 0 开始的，所以 i = integerPart + 1

3. 当 decimalPart  = 0 时，分位数 = x[i]

4. 当 decimalPart != 0 时，分位数 = x[i] + decimalPart * (x[i+1] - x[i])

如果忽略 dimsSortId 的存在，得到一个新的 CBitmap 结构：

CBitmap:

Map(short → Map(dimsSortId →  RoaringBitmap(uid)))

转化为 Map(short → RoaringBitmap(uid))

之前生成事件指标的 Cbitmap 如下：

{

1 → {0 → (1), 1 → (1)},

0 → {0 → (2, 3, 4)}

}

转化后的 CBitmap 如下：

{

1 → (1),

0 → (2, 3, 4)

}

Bitmap 分位数到底咋算

🤔 给点数据，给个需求，先来个简单的，数据如下：

cBitmap = { 3 -> (1001, 1006) 2 -> (1003, 1005, 1006) 1 -> (1004) 0 -> (1001, 1002, 1003)}求这个CBitmap的75分位数？

🤔 把每个用户对应的 cnt 按照顺序拿出来，再按照公式求分位数？

cBitmap转成cnt从小到大排序后的映射关系(uid -> cnt)【  (1002 -> 1),   (1004 -> 2),  (1005 -> 4),  (1003 -> 5),  (1001 -> 9),  (1006 -> 12)】
75分位数求解:x = (1, 2, 4, 5, 9, 12)(6 - 1) * 0.75 = 3.75分位数 = x[i] + decimalPart * (x[i+1] - x[i]) = 5 + 0.75 * (9 - 5) = 8

看似木得问题，实则慢得一匹。。。

从上到下依次遍历每一个 C 位获取每一个用户对应的 cnt，得到 cnt 的排序数组，最后再根据公式才能求出结果。

🤔 简单点儿，求解的方式简单点儿？

既然 CBitmap 本身就是计次且有序的，为啥不充分利用起来？

对于 cbitmap 求分位数，前提就是获取排序后第 i 个人和第 i + 1 个人对应的 cnt 数

① 计算终究需要知道 integerPart 是多少？

（6 - 1） * 0.75 = 3.75

i = 3 且有小数部分，需要获取 x[i] 和 x[i+1]

② 由于 cbitmap 中高位的数据一定比低位的数据大，所以 cbitmap 计算第 i 个人可以从高位开始遍历排除数据，拿到第 i 个人的 c 位

• 先记录几个必要的变量

totalRbm = cbitmap 去重后用户集合

currIdx = 当前遍历的 c 位

currRbm = 当前指针位置对应的 roaringBitmap

persistRbm = 一定是小于当前指针位置的这部分用户 = totalRbm andNot currRbm

preDiscardRbm = 本次遍历准备排除的高位的用户 = totalRbm and currRbm

cBitmap = { 3 -> (1001, 1006) 2 -> (1003, 1005, 1006) 1 -> (1004) 0 -> (1001, 1002, 1003)}
totalRbm = (1001, 1002, 1003, 1004, 1005, 1006)currIdx -> currRbm = 3 -> (1001, 1006)persistRbm = (1002, 1003, 1004, 1005)preDiscardRbm = (1001, 1006)

• 排除算法，找到第 i 个用户的 c 位

1. 如果 persistRbm 的人数 >= i，说明我们要找的第 i 个人还在 persistRbm 中，并将 persistRbm 置为 totalRbm， currIdx -= 1，重新计算那几个重要变量进入排除算法。

2. 如果 persistRbm 的人数 < i，说明第 i 个人就在当前的 preDiscardRbm 中，此时需要一个 resCnt 记录当前 currIdx 对应的 count 值【1 << currIdx】，如果之前有值，则需要累加。

1) 下一步是在 perDiscardRbm 中找到需要的那个人，新的位置 =  之前的 i - persistRbm 人数。

2) 并将 preDiscardRbm 置为 totalRbm， currIdx -= 1，重新计算重要变量进入排除算法。

3. 遍历完 c 位，即 currIdx = 0 时，累加的 resCnt 即为 x[i] 的结果。

• 来吧，演示：

• 当还需要获取第 i+1 个位置的用户是否还得重新遍历一次？

将接收变量换成数组，在排除算法中一次遍历获取第 i 个用户和第 i+1 个用户，只需要考虑以下两个情形：

1）当第 i 个用户和第 i+1 个用户在同一 cnt 位上，则后续排除算法的判断逻辑无差异。

2）当出现第 i+1 个用户在 currIdx，而第 i 个用户在 currIdx - 1 时，导致 totalRbm 不一致，需要分开进行计算。

• 最后整合 x[i] 和 x[i+1] 的结果，得到分位数：

分位数 = (x[i] + decimalPart * (x[i+1] - x[i])) * weight

总结

本篇主要揭晓基于 BitMap 来作为底层的数据模型实现的 Percentile 算法的优势，Bitmap 的高度压缩在存储方面带来巨大优势的同时，还能根据其数据结构灵活计算统计数据，快速计算许多类似 Percentile 的需求。

BitMap 还有很多的扩展性和亮点，下面列举几个，敬请期待：

• BitMap 黑科技：避免反序列化使用字节进行 BitMap 运算

• BitMap 转置算法：不一样的 count 求解方式

• …

参考资料：

1. https://www.jianshu.com/p/e271caf2d4e2
   

2. https://xiaoyue26.github.io/2019/05/26/2019-05/spark-sql中的分位数算法/
   

3. https://www.jmlr.org/papers/volume11/ben-haim10a/ben-haim10a.pdf
   

4. http://dx.doi.org/10.1145/375663.375670
   

关于 GrowingIO

GrowingIO 是国内领先的一站式数字化增长整体方案服务商。为产品、运营、市场、数据团队及管理者提供客户数据平台、广告分析、产品分析、智能运营等产品和咨询服务，帮助企业在数字化转型的路上，提升数据驱动能力，实现更好的增长。

点击「此处」，获取 GrowingIO 15 天免费试用！