Uber operates on a massive scale, facilitating over 2.2 billion trips every quarter. Deriving even simple insights necessitates a scaled solution. In our case, we needed to count the number of jobs someone had participated in while on the Uber platform, for arbitrary time windows. This article focuses on the challenges faced and lessons learned as we integrated Apache Pinot™ into our solution.

Specifically our solution needed to resolve:

Our previous solution was simple: retrieve jobs for a given subject with a page size limit of 50, and paginate the result until there are no further jobs. In Uber’s early days, with no single account accruing comparatively much tenure, this worked well. However, Uber ventured into new verticals, and some accounts began to present tenure in the tens of thousands, it became clear that we needed a more robust solution.

A main product requirement was that this solution must be able to compute tenure lookback. This by itself would have been tenable, but accompanied by our data retention policy, it was deemed unreasonable to accommodate by our downstream team.

The same team, in the interest of cost savings, determined that data older than 2 years would be sequestered into a higher latency storage tier. However, a change of plans mid-project resulted in them dropping online access to this data altogether. 

Our solution at the time required stitching three data sources for each of Uber’s marketplaces: Rides and Eats.

Immediately after presenting our design to the wider team, we found teams across Uber having adjacent projects with similar requirements, each reimplementing their own solution to count job tenure independently.

Taking these constraints into account, we considered several architectures. One we seriously considered was backed by Apache Hive™, and Docstore (Uber’s in-house distributed database), finally landing on a solution leveraging Apache Pinot™. An interesting feature here is the hybrid table, presenting a seamless interface which stitches real-time and offline data together.

Apache Pinot™ is an extremely powerful product, giving us unparalleled flexibility with our schema. However, we encountered several roadblocks along the way, and below, we detail the challenges and our solutions to these issues that may be insightful to the reader.

Our first challenge was to formulate a capacity requirement to provision the hardware required for our dedicated tenant. Apache Pinot™ utilizes compression techniques, making it difficult to measure space on disk a priori. Here we opted to take a 10% sample dataset and projecting disk usage based on the space taken by the sample was sufficient in our case.

One interesting side note here is that, while we were able to approximate dataset size on disk via sampling, we were unable to accurately predict query performance. In this case, we had to wait until after we scaled up for the entire dataset before we were able to acquire crucial metrics, e.g. p99 read time, disk read throughput, etc. Once we were able to do so, production-sized traffic brought our dedicated cluster to its knees, seeing read times exceeding 10s broker limits, maxing out read throughput on SSDs, and CPU usage. We proceeded to immediately investigate optimizations, as it was not unexpected, since each query was tantamount in load to a full table scan.

For reference, our query is of the shape:

This concerned the team, and we took several steps to improve performance, including:

As Uber continues to add features to the platform, so do their downstream impacts on source data. Currently, we can receive trip-level information via three modalities: Apache Hive™, Apache Kafka™ Topic, and API response, each complete with a different schema. Retrofitting and representing new features into an existing schema in a sensible way can be difficult, in particular, Apache Hive™ schemas. Historical data holds a lot of inertia and can make migrating schemas unreasonable.

For example, consider the Fare Split feature, where the cost of a ride can be split amongst multiple riders. Prior to this feature, it was always true that a job had one rider, which was always the payer, and each Hive record implied one job performed by the driver on the order. These invariants no longer hold true. The choice was made here to duplicate records and set the status to FARE_SPLIT, while setting driver_uuid column to NULL.

It is complications like this that make performing a simple COUNT DISTINCT aggregation not possible. For each business case, deciding whether or not a record contributes to a subject’s tenure must be clearly thought out:

Another challenge we had to solve here was slow upstream data. While most data arrives in seconds, it’s possible that a trip may not appear in upstream data sources for up to a week. To solve this, we created a pipeline that dynamically generates backfill pipelines, but only scheduled to run between T – 7d and T.

Alongside this, we also perform offline data quality checks which are simple COUNT(*) queries for T – 1d date to ensure our source Apache Hive™ tables are in sync with results from our Apache Pinot™ hybrid table.

Upstream traffic with spiky traffic patterns was also an issue. In this case, the specific rate limiting implementation caused a large spike in traffic every ten seconds, each of which was breaching the ten-second Apache Pinot™ broker timeout, failing the request.

We addressed this by adding jitter to our upstream client at request time to more evenly distribute our queries over time.

After overcoming these challenges, we have been serving live production traffic for almost a year, and performing load testing shows at least 200% headroom, after including a buffer for failover traffic. Our p99 read latency is ~1s: impressive, as some of our upstream queries can hit over 2,000 segments, with each segment consuming approximately 90 MB of disk space. After iterating on our solution, we began to publish metrics comparing our previous downstream with our solution. Our Apache Pinot™-based solution presented an almost 1:1 accuracy, giving us the confidence to rely on it. We first gated the change with a config flag, and after some time, cut over fully and deprecated the previous implementation.

With some simple additions, we are confident we will be able to answer more powerful questions. For example:

Finding point-in-time tenure at job granularity is expensive. Avenues to improve performance and reduce storage costs here are still being explored. While still in the design process, we expect that by dropping the job-granularity requirement, we should be able to significantly increase read throughput.

Special thanks to Caner Balci, Qiaochu Liu, Jacob Sevart, and Ujwala Tulshigiri for their contributions to this post.