In today’s fast-paced world of software development, new changes (both in code and infrastructure) are being released at breakneck pace. At Uber, we roll out ~11,000 changes every week and it’s important for us to have a way to quickly be able to identify and resolve issues caused by these changes. A delay in detecting issues can create a number of issues including impacts to: user experience, our ability to facilitate transactions on the platform, company revenue, and the overall trust and confidence of our users. At Uber, we have built a system called “Healthline” to help with our Mean Time To Detect (MTTD) and Mean Time To Resolve (MTTR) issues and to avoid potential outages and large-scale user impacts. Due to our ability to detect the issues in real time, this has become the go-to tool for release managers to observe the impact of canary release and decide whether to proceed further or to rollback. 

In this article we will be sharing details on how we are leveraging Apache Pinot^™ to achieve this in real time at Uber scale.

Healthline is a crash/exception logging, monitoring, analysis, and alerting tool built for all Uber mobile applications for multiple platforms as well as more than 5,000 microservices written using various programming languages, and owned by a number of tech teams within Uber. 

On a very high level, it processes all the crashes, errors, and exceptions logs generated by internal systems and builds analytical insights on these by classifying the similar exceptions/crashes into different buckets called issues. It also solves for identifying potential changes causing these issues and subsequently notifying the system owners whenever there is an anomaly observed for a specific issue. 

For this article, we will limit our scope to Mobile App Crash Analytics, however it’s quite similar for backend logs. Also we will not be covering the data collection pipelines and preprocessing steps like deobfuscation of crash dumps, classification logic, code change, and ownership identifications.

Crash – We are using it to represent an instance of any fatal or non fatal (e.g., memory leak) report we receive from our SDK embedded in each Uber mobile app.

Issue – A cluster of crashes which are identified as having the same root cause.

App – An app represents an Uber-developed app. We consider App + Platform (iOS/Android) as one unique app, hence Rider iOS and Rider Android are counted as two apps.

At peak, we classify more than 1.5 million errors and warning logs from backend services, and more than 1,500 mobile app crashes per second across Uber apps for both iOS and Android. The size of the crash varies from 10 KB to 200 KB with an average per-day data size of 36 TB. The crash QPS is ~1,500. We retain the data for 45 days to understand historical trends. Most of our use cases access the last 30 days data.

For dashboards and to perform alerting we perform following queries with expected response time:

We have a large number of attributes on which aggregation can be performed, and they may occur together in the same query.

Following are the aggregation patterns identified (with simplified examples):

We also display a variety of graphs/histograms on the UI. They can be generalized as “for a given time window, divide it into equal length bins, and for each bin, perform a particular aggregation.”

It’s clear from above query patterns that we have a very strong use case for aggregations and matches (including range match). Some partial match use cases are present, however they are very low in QPS. More importantly, most of our queries have a time window specified. This makes Pinot tailor-fit for our use case, with its rich support for various types of indices. We also have a strong platform team providing managed Pinot offering. Pinot also has a very strong open source community with some of the committors from Uber.

Apache Pinot is a real-time, distributed, columnar OLAP datastore, which is used to deliver scalable real time analytics with low latency. It can ingest data from batch data sources (such as Apache Hadoop^® Distributed File System, S3, Azure^® Data Lake, Google Cloud Storage) as well as streaming sources (such as Apache Kafka^®). Pinot is designed to scale horizontally, so that it can scale to larger data sets and higher query rates as needed.

A table is a logical abstraction that represents a collection of related data. It is composed of columns and rows (known as documents in Pinot). The columns, data types, and other metadata related to the table are defined using a schema.

Pinot breaks a table into multiple segments and stores these segments in a deep-store such as Hadoop Distributed File System (HDFS) as well as Pinot servers. Pinot supports the following types of tables:

Excerpt taken from Apache Pinot Official Documentation. For more information, please visit the Apache Pinot Docs.

Our ingestion pipeline processes raw crash data and enriches it with classification and other details. It publishes this enriched crash data to a Kafka topic. This is the source which we want to store and want to perform required aggregations and filtering operations to serve to user dashboards and other integrations.

Hybrid Pinot tables have both real-time as well as offline tables under the hood.

Offline tables ingest pre-built Pinot segments from external data stores and are generally used for batch ingestion. Real-time tables ingest data from streams (such as Kafka) and build segments from the consumed data.

We utilize both real-time and offline tables for maximum efficiency as well as cost saving. As we can see in the diagram, there is a data overlap between the tables. This is intentional and needed. It is done to hedge against the fact that if the offline job experiences failures in future, the overlapping data is still present in the real-time table and we avoid complete data loss.

Although, while querying in case of overlap, Pinot gives higher priority to offline data and fetches from there.

Since our use case is primarily that of aggregation, Pinot is the star candidate. We use Pinot to store data in various schemas, and perform aggregations to surface the metrics on UI.

The crash events received by Healthline follow a highly nested structure, consisting of primitive and collection objects. Since Pinot doesn’t support storing nested objects, we flatten the fields before dropping a crash event in Pinot. Also, since the bulk of crash events consists of the crash dump, and we only need to perform aggregations on metadata of the event, we store a subset of fields by flattening them. The whole crash event is also compressed and stored in a separate Pinot table which the users can retrieve on demand to view the crash dump.

We have implemented annotation-based data flattening and compression.

Since Pinot does not allow “.” in the column name, we converted it to “_” and to keep naming consistent, switched to snake case.

Our entity contains the fields experiments and analyticsLogs, which are an array. We perform aggregations on this array and surface the analytics for individual array elements on healthline. Since Pinot does not allow querying/aggregation on array elements, we took the approach to explode the data. 

For each issue ID, get the array field of size n, and create n rows of data, each containing issue ID, and array element [i]. We store this denormalized data in healthline_crash_event_denormalized_prod. This table contains an additional column called “type” that indicates to which field the record corresponds.

Crash Events processed by healthline are huge payloads, varying from 10 kB to 200 kB per event. Even after compressing the raw crash events, depending on the compression ratio achieved using the GZip compression algorithm, compressed events vary till 100 kB. The entire raw crash event is primarily used for the below use cases:

Both the above use cases are essentially a primary key (i.e.. crash_uuid) based lookup. Both the write and read path access patterns can be solved by a row-key-based lookup. The right database choice in such a scenario would have been a NoSql/Blob store, which can solve these use cases in a better way.  However to avoid the complexity of additional DB integration and maintenance (infrastructure and development) cost, it is decided that Pinot will be used as a blob store in this case, even though it is not a right use case supported by Pinot. 

During the POCs it is identified that segment size for a single table containing all the report types is getting bigger. Some of the data points around this are:

On a business-as-usual day, a single table containing all the crash types for 6 hours has a total table size of 40 GB and total of 30 Segments with an average size of 1.1 GB per segment. For each segment, no records are 100K. These numbers are not ideal for a regular Pinot table, and because of this there can be reliability issues. 

To solve this (data size) problem there can be multiple solutions that need to be explored: 

Solution 1 –  Reduce the Retention time for both the tables (Offline and Real-time). Reducing the TTL might help in reducing the segments for offline tables, but not for real-time tables. At the current throughput, real-time tables will still have higher size per segment. 

Solution 2 – GZIP compression algorithm is providing compression ratios of 50-60%. Using any alternative compression algorithm, even if we achieve 20% more compression, it will still not make the Pinot table reliable, as segment size would still be on the higher side.  

Solution 3 – Split the data into multiple tables: 

Solution 4 – Apply random sampling on the data which is leading to higher data growth (discussed in next section).

*sampling – By sampling in this document, we mean sampling the compressed data only. We would still be recording and storing index fields for ALL crashes, as the aggregation numbers should not be affected.

We display sample crash instances to users on the UI. Since a human user could only sift through a limited number of instances, it makes sense to store only a subset of compressed data. Currently the system shows up to 1,000 instances, which can safely be brought down to a couple hundred.

A point to note is that we do not want a very small time window, as it will create a lot of buckets and Flink will spend too many resources on managing and distributing the buckets, making the job less efficient. Hence it is a tradeoff between resource management overhead and job stability. By POCs, 10s time window was found to suit our use case.

A config is created which holds the mapping between flattened and JSON keys along with class type. For example, for the nested JSON integer node A.B.C, the config will hold:

A_B_C -> A.B.C, Integer

This is a one-time operation which uses reflection to construct the config map. Using this config, we can quickly determine the JSON node of a given class attribute.

Since there is a bug on Neutrino which converts bytes array to string inconsistently, we decided to do the string conversion at application level and store the compressed_data_string in Pinot. The bytes-to-string conversion uses a base-64 encoder. To decompress compressed_data_string:

Pinot supports a variety of indexes. However for our use case (filter patterns on data), we required 3 types of indexes:

We have migrated from using Elasticsearch to Pinot. The following table represents Elasticsearch vs. Pinot performance comparison when data was queried over a range of time periods. Time duration is in seconds.

The performance degradation in Elasticsearch is much higher compared to that of Pinot when the time period is increased.

Since we have migrated from using Elasticsearch to Pinot, the following numbers represent the impact of said migration:

Using Pinot for crash analytics has proven faster, cheaper, and easier than alternatives. This has certainly enhanced the user experience while bringing down operational costs for the Healthline team. We have also managed to unlock better analytical capabilities, which were earlier not possible or readily available. We are already working on next-gen healthline features like Breadcrumbs (a trail of console/memory/network logs for a given issue) and mobile crash- to-backend logs correlation.