在分布式数据管道中扩展高基数非汇总指标的精确 COUNT(DISTINCT)

简介

Uber 本季度服务了多少独立用户？对于这类任务关键型指标，标准的工程答案是近似值。我们需要精确计数，但在我们的规模下，此前并无现成方案。

在 Uber，有一类指标——如月活跃用户、季度留存率和跨窗口参与度——必须精确计算。这些是非汇总（non-rollup）指标：它们的值无法从更粗粒度的预聚合中推导出来，日计数或月计数的任何组合都无法得出正确的季度结果。每个独立实体都必须在整个聚合窗口内被同时追踪。在季度规模下，这意味着需要在内存中同时持有超过 36 亿个 UUID 值的标识符。

RoaringBitmap 是此类问题的行业标准：紧凑、快速且精确。但在基数足够高时，它会撞上 JVM 中的一个硬性限制：Java 的内部数组索引受 Integer.MAX_VALUE（约 2 GB）限制。当 reduce 阶段完全合并后的位图超过此限制时，作业每次运行都会确定性地以 OutOfMemoryError 终止，且没有任何配置可以规避。社区的答案是改用近似计数。我们不能这么做。

为了解决这个问题，我们构建了一种分块聚合缓冲区策略：重新设计位图聚合缓冲区，将其按映射（map）分区，键为每个哈希标识符的高 16 位。由于每个分块可以独立序列化，我们将峰值内存限制在最大的单个分块内，从而彻底消除了 2 GB 的 JVM 限制。该方案已在 Uber 的 75 个指标族中部署，使两年期回填时间平均缩短了 65%，并在部署后消除了所有内存溢出故障。

如果你运行基于 JVM 的数据管道，在 Apache Hive™、Apache Spark™ 或任何需要将聚合状态作为单个 byte[] 返回的框架上，大规模计算精确去重计数，那么这个问题离成为你的问题只差一个基数阈值。本篇博文将介绍我们如何解决这个问题，以及为什么该方案可以推广到 Uber 技术栈之外。

非汇总指标的挑战

大规模数据平台中的聚合指标分为两类。汇总（rollup）指标（如完成行程数）可以跨时间窗口干净地组合：周总计就是日总计之和。这些指标可以在更细的粒度上预聚合，之后再进行组合。非汇总指标则不然。MAU（月活跃用户）是一个典型例子：一个在周一和周三都活跃的用户，在周总计中必须只计一次，而不是两次。日活跃用户计数的任何组合都无法得出正确结果。唯一正确的计算方式是在整个窗口内同时追踪每个独立用户， 而在季度规模下，这意味着要在内存中同时持有 36 亿个 UUID 值的标识符。

图 1：汇总与非汇总指标聚合。 

如图 1 所示，汇总指标（完成行程）可跨粒度进行加法组合，而非汇总指标（MAU）则要求在整个聚合窗口内完整保留身份。对日计数进行简单求和会重复计算出现在多天的用户。

压缩位图索引代表了标准的架构解决方案：将集合成员关系编码为单个位，并使用高性能按位 OR 进行集合并集，以消除跨网络传输原始标识符的高昂成本。挑战在于，在这个特定的基数下，每种常规方法都会确定性地失败，每种方法都受阻于不同的技术约束。

图 2：非汇总指标计算的规模（评估时的 Uber 生产数据）。

Bitmap-32 与全局字典

RoaringBitmap 处理 32 位整数，需要通过共享的全局字典将 UUID 映射到唯一的整数秩。这种架构造成了一个关键的单点故障：字典中的任何损坏或复制延迟都会使所有进行中的作业失效。此外，并发作业会产生重叠的秩分配，这要求回填作业必须顺序执行。在我们的规模下，这使得两年历史数据的回填超过了 20 天。

Bitmap-64 与整体序列化

转向 Roaring64Bitmap 消除了对字典的需求，直接将 UUID 哈希为 64 位整数，冲突风险可忽略不计。然而，Hive 和 Spark UDAF 契约要求聚合状态作为单个 Java byte[] 返回，而 JVM 将其上限限制在约 2 GB。对于高基数 UUID 工作负载，这一限制在大约 1.79 亿个唯一标识符时就会被突破。 在 36 亿的季度规模下，这会导致每次运行都确定性地作业失败。

近似计数

HyperLogLog 绕过了这些限制，但引入了 1-5% 的误差，这对于支撑财务报告的 MAU 等指标来说是不可接受的。我们需要一个精确的、无字典的、且永远不会被物化为单个序列化对象的解决方案。当时并不存在这样的方案。

突破 JVM 屏障

有时候，解决方案不是通过跳出思维定式找到的，而是通过利用更多的盒子找到的。

JVM 架构施加了一个硬性限制，拒绝分配任何超过 2 GB 的单个数组。我们没有使用整体的 Roaring64Bitmap，而是实现了一个 Map<Integer, Roaring64Bitmap>，其中每个条目处理 64 位值空间的一个不相交段。由于这些切片不重叠，精确的总数只需将每个切片的基数相加即可得出。这就是分块聚合缓冲区策略，如图 3 所示。

图 3：64 位哈希值的 16 位块分区。 

高 16 位通过 chunkId = (int)(hash >>> 48) 决定路由到特定的块标识符。然后将完整的 64 位值持久化到相应块的 Roaring64Bitmap 缓冲区中。

每个传入的 UUID 都通过 Hive 原生哈希函数 xxHash64 转换为 64 位整数。然后哈希的高 16 位决定路由到特定的块： 

chunkId = (int)(hash >>> 48)

我们选择 xxHash64 有两个原因：作为 Hive 内置函数，它不需要任何外部依赖。其雪崩特性也保证了在生产 UUID 工作负载中，所有 65,536 个块 ID 的均匀分布。

为什么是 16 位？ 

使用 16 位可将 64 位值空间划分为 65,536 个不同的块。在每季度 36 亿标识符的规模下，我们哈希函数的稳健分布特性使得每个块平均包含 55,000 个值。通过 Chernoff 界进行的统计分析证实，任何单个块达到 1.79 亿值 OOM 阈值的概率实际上为零。

较短的位深度提供的余量不足：8 位产生 256 个块，每个块平均约 1400 万个值，且无法保证避免最坏情况下的倾斜。较长的位深度（24 位）会产生超过 1600 万个可能的块，不成比例地增加映射开销和 GC 压力。

解锁并发执行

Bitmap-32 全局字典迫使回填作业顺序运行，以防止排名分配冲突。通过直接哈希到 64 位整数，我们消除了作业之间的所有共享状态；现在多个回填窗口默认并发执行。

自描述的部分结果

我们在每个分块部分结果前加上 4 字节的魔术数字：0xDEADBEEF。读取部分结果时，评估器会检查前 4 个字节；匹配则路由到分块合并路径，不匹配则路由到传统反序列化路径。选择 0xDEADBEEF 是有意为之：Roaring64Bitmap 的内部 cookie 头永远不会与此值冲突，使得误报在结构上不可能发生。

流式序列化

在部分聚合期间，每个块独立序列化到自己的缓冲区，并顺序写入输出流。通过将部分结果的大小限制为最大的单个块而非所有块的并集，我们在所有中间阶段有效地绕过了 2 GB JVM 限制。

仅基数的最终输出

在最终 reduce 阶段，terminate() 函数完全避免了位图序列化。它只是聚合已填充块的基数并返回一个 8 字节的 long。这确保了最终输出无论总去重计数如何都保持恒定的 8 字节，消除了 reduce 阶段的 OOM 风险。

总体而言，这些机制将我们的安全基数阈值从约 1.79 亿扩展到约 11.7 万亿，实现了 65,000 倍的可扩展性提升。

图 4：分块策略前后的有效安全基数阈值；虚线表示实际生产工作负载的基数。 

大规模零干扰迁移

我们采取了积极措施，在生产环境中实施此解决方案，且未造成任何干扰。 

与 Hive UDAF 生命周期的无缝集成

分块位图 UDAF 可清晰映射到 Hive 的四个生命周期阶段：PARTIAL1 （每 mapper 聚合）、PARTIAL2 （combiner 合并）、FINAL（reducer）以及 terminate()。在 map 和 combine 阶段，每个节点仅序列化其本地分区，即完整值空间的一部分。在 FINAL 阶段，terminate() 完全绕过位图序列化，返回一个恒定的 8 字节 long。

无对象物化的流式序列化

每个分块独立序列化到其自身的 ByteArrayOutputStream 中，并按顺序写入输出流。反序列化时，mergeFromStream() 从 DataInputStream 中一次读取一个分块，分配一个与分块大小完全匹配的 byte[] ，将其合并，并在处理下一个分块之前释放它，从而将峰值内存限制在最大的单个分块，而非总负载。

跨双路径的无缝迁移

我们的迁移涉及两种遗留架构。绑定到全局字典的 Bitmap-32 管道被迫顺序执行并引入了损坏风险，经过全新迁移，淘汰字典后实现了并发回填执行，且无需跨作业协调。

相反，一些管道已经在使用单体式 Bitmap-64，因为其基数仍低于 1.79 亿的 OOM 阈值；有效的 Roaring64Bitmap 部分结果已持久化在 HDFS 中。对于这些工作负载，自描述的魔数路由器促进了无缝互操作。遗留部分结果会被自动检测并通过原始反序列化路径路由，以合并到分块缓冲区中，而原生分块部分结果则通过优化的合并路径处理。这一结构性保证确保了无论属于哪种遗留类别，任何管道都不需要历史重算窗口。

严格正确性验证

在全面投入生产之前，分块 UDAF 生成的去重计数与 10 个采样指标族的原始 COUNT(DISTINCT identifier) 结果进行了交叉核对。这些基准测试涵盖了每日、每月和每季度管道，以确保在不同规模下的准确性。在所有 30 次验证运行中，结果完全匹配，零偏差。我们还针对单个作业内混合部分结果类型对路由器进行了压力测试，确认了路由的确定性。

Uber 的用例

分块聚合缓冲区策略 现已成为 Uber 出行、配送和平台各垂直领域精确 COUNT(DISTINCT) 的平台标准实现。自部署到 75 个指标族以来，生产环境中未观察到 OOM 故障。

回填时长

在历史数据重处理中，两年期回填的平均时长从 8.58 天降至 2.96 天，与顺序 Bitmap-32 基线相比减少了 65%。性能提升呈右偏分布，基数最高的管道提升幅度高达 94%。最引人注目的是，一个大规模的杂货参与度指标的回填时间从 19.1 天骤降至仅 1.1 天。这一加速由并发执行驱动：消除共享字典状态使得并行回填窗口成为可能，在 3 个并发窗口中实现了平均 3.8 倍的加速。

每日管道运行时长

每日管道的生产执行时间平均提升了 23%。这一效率提升源于 shuffle 数据量的减少：分块 UDAF 为每个分组键返回一个恒定的 8 字节 long 值，而非数千字节的位图，从而降低了网络 I/O 和 HDFS 写入开销。

每月数据准备

端到端的月度数据准备周期缩短了 43%，从 20 天降至约 10 天。这一提升既反映了运行时性能的改善，也得益于消除了因 OOM 导致的重新执行——此前在数据累积期间，这类问题常常迫使整个流水线重启。

后续步骤 

解决序列化瓶颈打开了更广阔的架构变革空间。随着精确的高基数聚合如今在大规模场景下变得可靠，非汇总指标可以从孤立的批处理流水线升级为一等的交互式原语，团队可以实时切片和筛选精确的去重计数，而无需牺牲准确性。以下是三个具体方向：

• 扩展非汇总指标的分析能力。随着规模限制的解除，去重计数的非汇总指标现在可以支持更丰富的分析操作、维度切片、群体对比和趋势分析——这些此前因 OOM 上限而无法实现。
• 自助式指标原语。 将分块精确 COUNT(DISTINCT) 作为一等类型暴露在 Uber 的 uMetric 平台中，使任何团队都能定义高基数非汇总指标，而无需编写自定义 UDAF 代码或承担 OOM 风险。
• 超越 UUID 工作负载的泛化。 该分区策略适用于任何 64 位哈希域——设备 ID、会话令牌、geohash 编码的位置——将精确的高基数聚合扩展到更广泛的保持身份标识的度量类别。

结论

在季度基数规模下扩展精确的 COUNT(DISTINCT) 需要面对一个任何配置更改、内存调优或框架升级都无法解决的限制。JVM 的 2 GB 数组限制是结构性的，唯一的出路就是结构性的解决方案。

分块聚合缓冲区策略将聚合缓冲区划分到互不相交的值空间段中，对每段独立序列化，并将基数作为 8 字节 long 返回，在每个阶段消除了序列化瓶颈，无需更改 schema、协调重算，也不牺牲任何精确性。一个魔数哨兵值确保了在整个迁移过程中与遗留流水线数据的无缝共存。

该实现部署在覆盖 Mobility、Delivery 和 Platform 的 75 个指标族中，回填时长平均缩短了 65%（最高基数的工作负载缩短达 94%），每日运行时长改善了 23%，月度数据准备时间缩短了 43%。部署后未记录到任何 OutOfMemoryError 失败，在数十亿个 UUID 值标识符上保持了精确去重计数的准确性。

致谢 

我们衷心感谢 Spark 团队提供的配置指导和更深入的平台见解，以及 MTD GSS 团队在数据验证和整个 Chunked Bitmap UDAF 迁移过程中给予的支持。

封面图片出处： “Vintage Wooden Mailbox with Letters Inside”，作者 Hassan Bouamoud，可免费使用。

Apache、Apache Spark、Spark、Apache Hive 和 Hive 是 Apache Software Foundation 在美国和/或其他国家/地区的注册商标或商标。使用这些标记并不意味着 Apache Software Foundation 的认可。

Java、MySQL 和 NetSuite 是 Oracle® 及其关联公司的注册商标。