Introduction

How many unique users did Uber serve this quarter? For mission-critical metrics like these, the standard engineering answer is approximate. We needed an exact count, but no prior solution existed at our scale.

At Uber, a class of metrics such as monthly active users, quarterly retention, and cross-window engagement must be computed exactly. These are non-rollup metrics: their values can’t be derived from coarser pre-aggregates, and no composition of daily or monthly counts yields a correct quarterly result. Every unique entity must be tracked simultaneously across the full aggregation window. At a quarterly scale, that means holding over 3.6 billion UUID-valued identifiers in memory concurrently.

RoaringBitmap is the industry standard for this class of problems: compact, fast, and exact. But at high enough cardinality, it hits a hard constraint baked into the JVM: Java’s internal array indexing is bounded by Integer.MAX_VALUE (of around 2 GB). When the fully merged bitmap at the reduce stage exceeds this, the job terminates with OutOfMemoryError deterministically, on every run, with no configuration workaround. The community’s answer has been to switch to approximate counting. We couldn’t.

To solve this, we built a chunked aggregation buffer strategy: a redesign that partitions the bitmap aggregation buffer across a map, keyed on the top 16 bits of each hashed identifier. Because each chunk serializes independently, we bound peak memory to the largest individual chunk and eliminated the 2 GB JVM constraint entirely. Deployed across 75 metric families at Uber, it reduced two-year backfill time by 65% on average and eliminated all out-of-memory failures post-deployment.

If you run JVM-based data pipelines computing exact distinct counts at scale on Apache Hive™, Apache Spark™, or any framework where aggregation state must be returned as a single byte[], this problem is one cardinality threshold away from being yours. This blog walks through how we solved this problem and why the solution is generalizable beyond Uber’s stack.

Challenges with Non-Rollup Metrics

Aggregate metrics in large-scale data platforms fall into two categories. Rollup metrics like completed trip counts compose cleanly across time windows: a weekly total is simply the sum of daily totals. These can be pre-aggregated at finer granularities, and combined later. Non-rollup metrics don't work this way. MAU (Monthly Active Users) is a defining example, a user active on both Monday and Wednesday must be counted once in the weekly total, not twice. No combination of daily active user counts produces the correct result. The only way to compute it correctly is to track every unique user simultaneously across the full window,  and at a quarterly scale, which means holding 3.6 billion UUID-valued identifiers in memory at once.

Figure 1: Rollup vs. non-rollup metric aggregation. 

As Figure 1 shows, rollup metrics (completed trips) compose additively across granularities and  non-rollup metrics (MAU) require full identity preservation across the aggregation window. Naive summation of daily counts overcounts users appearing on multiple days.

Compressed bitmap indexes represent the standard architectural solution: encoding set membership as individual bits and using high-performance bitwise OR for set unions to eliminate the prohibitive cost of shuffling raw identifiers across the network. The challenge is that at this specific cardinality, every conventional approach fails deterministically, each breaking against a different technical constraint.

Figure 2: Scale of non-rollup metric computation (Uber production figures at time of evaluation).

Bitmap-32 and the Global Dictionary

RoaringBitmap processes 32-bit integers, requiring a mapping from UUIDs to unique integer ranks through a shared global dictionary. This architecture creates a critical single point of failure: any corruption or replication lag in the dictionary invalidates all in-flight jobs. Furthermore, concurrent jobs generate overlapping rank assignments, mandating that backfill jobs execute sequentially. At our scale, this pushed two-year historical data backfills to over 20 days.

Bitmap-64, Monolithic Serialization

Transitioning to Roaring64Bitmap removes the dictionary requirement by hashing UUIDs directly to 64-bit integers with negligible collision risk. However, the Hive and Spark UDAF contract mandates that aggregation state be returned as a single Java byte[], which is capped at around 2 GB by the JVM. For high-cardinality UUID workloads, this limit is breached at approximately 179 million unique identifiers. At a quarterly scale of 3.6 billion, this results in deterministic job failure on every run.

Approximate Counting

HyperLogLog bypasses these constraints but introduces a 1-5% error, unacceptable for metrics like MAU that underpin financial reporting. We needed a solution that was exact, dictionary-free, and never materialized into a single serialized object. None existed.

Breaking the JVM Barrier

Sometimes the solution isn’t found by thinking outside the box, it's found by utilizing more boxes.

The JVM architecture imposes a hard limit, refusing to allocate any single array exceeding 2 GB. Instead of a monolithic Roaring64Bitmap, we implemented a Map<Integer, Roaring64Bitmap> where each entry handles a disjoint segment of the 64-bit value space. Because these slices are non-overlapping, the exact total is derived simply by summing per-slice cardinalities. This is the chunked aggregation buffer strategy, as shown in Figure 3.

Figure 3: 16-bit chunk partitioning of 64-bit hashed values. 

The top 16 bits dictate the routing to a specific chunk identifier via chunkId = (int)(hash >>> 48). The full 64-bit value is then persisted within the corresponding chunk’s Roaring64Bitmap buffer.

Each incoming UUID is transformed into a 64-bit integer via xxHash64, the native Hive hash function. The hash’s top 16 bits then dictate the routing to a specific chunk: 

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

We chose xxHash64 for two reasons: as a Hive built-in, it requires no external dependencies. Its avalanche properties also guarantee uniform distribution across all 65,536 chunk IDs for production UUID workloads.

Why 16 Bits? 

Using 16 bits partitions the 64-bit value space into 65,536 distinct chunks. At a quarterly scale of 3.6 billion identifiers, the robust distribution properties of our hashing function yield an average of 55,000 values per chunk. Statistical analysis via the Chernoff bound confirms that the probability of any single chunk reaching the 179 million-value OOM threshold is effectively zero.

Shorter bit depths offer insufficient headroom: 8 bits yields 256 chunks averaging around 14 million values each, with no guarantee against worst-case skew. Longer bit depths (24 bits) create over 16 million possible chunks, inflating map overhead and GC pressure disproportionately.

Unlocking Concurrent Execution

The Bitmap-32 global dictionary forced backfill jobs to run sequentially to prevent rank assignment collisions. By hashing directly to 64-bit integers, we eliminate all shared state between jobs; multiple backfill windows now execute concurrently by default.

Self-Describing Partial Results

We prefix every chunked partial result with a 4-byte magic number: 0xDEADBEEF. Upon reading a partial, the evaluator inspects the initial 4 bytes; a match routes to the chunked merge path, a mismatch to the legacy deserialization path. The choice of 0xDEADBEEF is deliberate: Roaring64Bitmap's internal cookie header can never collide with this value, making false positives structurally impossible.

Streaming Serialization

During partial aggregation, each chunk serializes independently into its own buffer and writes sequentially to the output stream. By bounding the size of the partial result to the largest individual chunk rather than the union of all chunks, we effectively bypass the 2 GB JVM constraint at all intermediate stages.

Cardinality-Only Final Output

In the final reduce stage, the terminate() function avoids bitmap serialization entirely. It simply aggregates the cardinalities across populated chunks and returns an 8-byte long. This ensures the final output remains a constant 8 bytes regardless of the total distinct count, eliminating OOM risks at the reduce stage.

In aggregate, these mechanisms expand our safe cardinality threshold from around 179 million to approximately 11.7 trillion, representing a 65,000× scalability improvement.

Figure 4: Effective safe cardinality threshold before and after the chunked strategy; dotted lines show actual production workload cardinalities. 

Zero-Disruption Migration at Scale

We took proactive measures to implement this solution in production without any disruption. 

Seamless Integration with the Hive UDAF Lifecycle

The Chunked Bitmap UDAF maps cleanly onto Hive's four lifecycle phases: PARTIAL1 (per-mapper aggregation), PARTIAL2 (combiner merge), FINAL (reducer), and terminate(). During the map and combine phases, each node serializes only its local partition, a fraction of the full value space. At the FINAL stage, terminate() bypasses bitmap serialization entirely, returning a constant 8-byte long.

Streaming Serialization Without Object Materialization

Each chunk serializes independently into its own ByteArrayOutputStream and writes sequentially to the output stream. During deserialization, mergeFromStream() reads one chunk at a time from a DataInputStream, allocates a byte[] exactly matching the chunk size, merges it, and releases it before processing the next, bounding peak memory to the largest individual chunk, not the total payload.

Seamless Migration Across Dual Paths

Our transition addressed two legacy architectures. Bitmap-32 pipelines, tethered to a global dictionary that forced sequential execution and introduced corruption risks, underwent a fresh migration retiring the dictionary enabled concurrent backfill execution with no cross-job coordination required.

Conversely, some pipelines were already using monolithic Bitmap-64, as their cardinalities remained below the 179 million OOM threshold; valid Roaring64Bitmap partial results were already persisted in HDFS. For these workloads, the self-describing magic-number router facilitated seamless interoperability. Legacy partials are automatically detected and routed through the original deserialization path to be merged into the chunked buffer, while native chunked partials are processed via the optimized merge path. This structural guarantee ensured that no pipeline, regardless of its legacy category, necessitated a historical recompute window.

Rigorous Correctness Validation

Prior to full production rollout, distinct counts generated by the chunked UDAF were cross-referenced against raw COUNT(DISTINCT identifier) results across 10 sampled metric families. These benchmarks covered daily, monthly, and quarterly pipelines to ensure accuracy at varying scales. Across all 30 validation runs, results matched exactly with zero deviation. We also stress-tested the router against mixed partial types within a single job, confirming deterministic routing.

Use Cases at Uber

The chunked aggregation buffer strategy is now the platform-standard implementation for exact COUNT(DISTINCT) across Uber's Mobility, Delivery, and Platform verticals. Since deployment across 75 metric families, zero OOM failures have been observed in production.

Backfill Duration

Across historical data reprocessing, the mean two-year backfill duration dropped from 8.58 days to 2.96 days representing a 65% reduction compared to the sequential Bitmap-32 baseline. Performance gains are right-skewed, with highest-cardinality pipelines achieving up to a 94% improvement. Most notably, a high-scale grocery engagement metric saw backfill time collapse from 19.1 days to just 1.1 days. This speedup is driven by concurrent execution: eliminating shared dictionary state allows parallel backfill windows, delivering a mean 3.8× speedup across 3 concurrent windows.

Daily Pipeline Runtime

Production execution time for daily pipelines improved by 23% on average. This efficiency stems from reduced shuffle volume: the chunked UDAF returns a constant 8-byte long per group key instead of multi-kilobyte bitmaps, lowering network I/O and HDFS write overhead.

Monthly Data Preparation

The end-to-end monthly data preparation cycle saw a 43% reduction, falling from 20 days to approximately 10 days. This gain reflects both improved runtime and the elimination of OOM-induced re-executions that previously necessitated full pipeline restarts during accumulation.

Next Steps 

Solving the serialization bottleneck opens a broader architectural shift. With exact high-cardinality aggregation now reliable at scale, non-rollup metrics can graduate from isolated batch pipelines into first-class, interactive primitives where teams can slice and filter exact distinct counts in real time, without sacrificing accuracy. Three concrete directions follow:

• Expanding analytics on non-rollup metrics. With the scale constraint removed, count distinct non-rollup metrics can now support richer analytical operations, dimensional slicing, cohort comparisons, and trend analysis that were previously impractical due to the OOM ceiling. 
• Self-serve metric primitive. Exposing chunked exact COUNT(DISTINCT) as a first-class type in Uber's uMetric platform, so any team can define high-cardinality non-rollup metrics without custom UDAF code or OOM risk.
• Generalization beyond UUID workloads. The partitioning strategy applies to any 64-bit hashed domain device IDs, session tokens, geohash-encoded locations extending exact high-cardinality aggregation across a broader class of identity-preserving measurements.

Conclusion

Scaling exact COUNT(DISTINCT) at quarterly cardinality required confronting a constraint that no configuration change, memory tuning, or framework upgrade could resolve. The JVM's 2 GB array limit is structural and the only way past it was a structural solution.

The chunked aggregation buffer strategy partitions the aggregation buffer across disjoint value-space segments, serializes each independently, and returns cardinality as an 8-byte long, eliminating the serialization bottleneck at every stage without schema changes, coordinated recomputes, or any sacrifice in exactness. A magic-number sentinel ensures seamless coexistence with legacy pipeline data throughout the migration.

Deployed across 75 metric families spanning Mobility, Delivery, and Platform, the implementation delivered a 65% mean reduction in backfill duration (up to 94% for the highest-cardinality workloads), a 23% improvement in daily runtime, and a 43% reduction in monthly data preparation time. Zero OutOfMemoryError failures have been recorded post-deployment, with exact distinct-count accuracy maintained across billions of UUID-valued identifiers.

Acknowledgments 

We extend our sincere thanks to the Spark team for their configuration guidance and deeper platform insights, and to the MTD GSS team for their support with data verification and assistance throughout the migration to the Chunked Bitmap UDAF.

Cover Photo Attribution:  “Vintage Wooden Mailbox with Letters Inside” by Hassan Bouamoud is free to use. 

Apache, Apache Spark, Spark, Apache Hive, and Hive are either registered trademarks or trademarks of the Apache Software Foundation in the United States and/or other countries. No endorsement by The Apache Software Foundation is implied by the use of these marks.

Java, MySQL, and NetSuite are registered trademarks of Oracle® and/or its affiliates.