Introduction

Presto^® is an open-source, distributed SQL query engine designed for running interactive analytic queries on data sources of any size, from gigabytes to petabytes.

At Uber, Presto is a critical engine for data analytics across various departments. The Operations team relies on it for dashboarding, while Uber Eats and marketing teams use its query results for pricing decisions. Presto is also essential to Uber’s compliance, growth marketing, and ad-hoc data analytics, making it a cornerstone of the company’s data-driven operations.

Uber operates around 20 Presto clusters across over 10,000 nodes in 2 regions, supporting approximately 12,000 weekly active users. These users run about 500,000 queries daily, reading around 100 PB from HDFS. Presto is used to query multiple data sources, including Apache Hive™, Apache Pinot™, MySQL^®, and Apache Kafka^®, through its extensible data source connectors.

This blog describes how Uber designed Presto express to reduce end-to-end SLA for fast-running Presto queries.

Problem Statement

Earlier last year, we observed Presto experiencing query slowness for multiple months. To work around this, we had to add more capacity. The problem of query slowness was caused by throttling. To keep ‌Presto clusters from getting overloaded, we have concurrency limits that limit the number of queries that can run concurrently on the cluster. This creates a fixed pipe, and all the queries have to contend for a spot in that pipe. 

For a query to get accepted for execution, it has to pass the user, consumer, cluster, and cluster group level concurrency checks. Incoming queries are queued up in Exeggutor post-validation and released to Prism for routing if they satisfy the checks. For a particular user or consumer, queries are processed in a first-come, first-served manner.

Figure 3 shows how many times user and consumer-level limits throttled scheduled queries over a 7-day period.

The cluster concurrency level limit also throttles queries. For example, throttling for batch low-tier queries based on cluster limits.

Figures 3 and 4 show the significant throttling that Presto queries have to go through.

However, looking at the Presto query latencies, we saw that P50 execution times of the queries were well under a minute. Here, the execution time only refers to the running time of the query in Presto and doesn’t include any queuing time for the query in Exeggutor.

That meant that out of the roughly 500,000 queries that Presto executed every day, more than half of the queries could easily finish in under a minute. To reduce the queue time of these fast-running queries, we designed a new approach.

Identifying Express Queries

We define an express query as any Presto query that can finish within 2 minutes. To identify express queries, we developed a method using historical data to predict whether an upcoming query is an express query. To do this, we tested the P90 and P95 query execution times using the exact fingerprint and abstract fingerprint of the query with lookback windows of 2 days, 5 days, and 7 days. An exact fingerprint is a query hash after removing comments and whitespaces. An abstract fingerprint also removes literal values from the query. An abstract fingerprint can identify similar queries coming from the same pipeline or data services even though the query isn’t the same.

We used this candidate definition to predict if a query was express: if the X runtime of the query in the last Y days based on Z fingerprint was less than 2 minutes. We explored variations where X was P90 or P95, Y was 2, 5, or 7, and Z was exact or abstract. 

This gave us 12 candidate definitions. To compare the various candidates, we defined Accuracy as True Positive/(True Positive +False Positive) and Coverage as (True Positive+False Positive)/ALL.

To test the accuracy of the prediction, we wrote a SQL query to analyze the historical data and come up with the prediction. In the analysis, we found that the P90 value of the abstract fingerprint with a 5-day lookback window provided the best accuracy and coverage, with values of 95.7% and 48.99%, respectively. So, we decided to use this query as our primary indicator for express queries moving forward. By implementing this optimized approach, we could better predict and identify express queries. We’ll continue to monitor and refine our model to ensure its ongoing accuracy and effectiveness.

We added a sink to ingest data from our query events topic to a Pinot table that can be queried in real time. Now when a query comes to Presto, we can determine whether that query is express using the Pinot query shown in Figure 9.

We use the _count to make sure that we make the prediction based on at least 5 previous runs of the query. If the percentile that we get above is less than 2 minutes and the count is greater than 5, then the query is deemed an express query. Since the Pinot table has the exact fingerprint and data for the last 90 days, we can easily change the method to use the exact fingerprint, a different percentile and/or larger or smaller lookback window.

One drawback of using historical data for prediction is that we won’t be able to predict accurately for new queries coming to the system. However, this isn’t a problem for batch workloads, where most of the queries are scheduled and repeat themselves over a set frequency.

The query prediction latency is very minimal, with most of the predictions being made in under 100 milliseconds.

Initial Express Design

In our initial design of the express feature, we designated one of our existing batch clusters to be an express cluster and added the aforementioned logic in Prism to determine if a query was express and should be sent to the express clusters.

However, this approach faced several issues:

Our initial attempt revealed crucial insights. Even though more than 50% of the low-tier batch queries were express queries, they consumed much less CPU. The resources allocated to express and non-express queries should match the CPU requirements of those queries. Beyond that, the slowdown experienced by the express queries couldn’t be solved just by the runtime isolation of the express queries from the non-express queries. ‌Express queries needed their own queues in the upstream systems (Exeggutor), so that an express query for a user or consumer didn’t get blocked by a non-express query for the same user or consumer sent earlier.

Final Design

In the final system that’s currently in production, we created a separate queue for express queries in Exeggutor that’s matched by a set of small express clusters. The express determination is done at the time of query validation itself, and the query is added to the express queue if applicable. The express queue has much higher user or consumer-level concurrencies, letting them run much more express queries than non-express queries.

Impact

Currently the express feature is only enabled for batch low-tier queries. The low-tier express clusters use around 10% of the total batch low-tier Presto resources, yet they run about 75% of the batch low-tier queries.

They also provide a much better SLA to ‌users. The p90 queuing time for express queries is about 10 seconds, whereas it can be in hours for non-express queries. 01:37:06 in Figure 16 refers to 1 hour, 37 minutes, and 6 seconds.

The p90 running time of the express queries is also obviously less, and the express queries provide a much better end-to-end SLA for users.

Next Steps 

For next steps, we want to look at separating the express cluster group. In our current implementation, express clusters are a sub-group of the major cluster groups like batch low tier, batch high tier, and interactive. The idea was that each cluster group would have its own set of express clusters. However, based on our experience in production,‌ express can be a separate cluster group instead of being a subgroup. So, an express query should be sent to this cluster group regardless of whether the query is a batch high tier, low tier or interactive query. The reasons for this are: 

So far, we have only differentiated between express and non-express queries, but we can have more classifications like small/medium/large and each classification going to a separate cluster group.

Another area we’re exploring is query cost prediction. Historical data has been our primary tool for predicting whether a query qualifies as an express query. Having historical data can be a challenge for interactive workloads. To address this, we can harness machine learning models to determine whether incoming queries should be classified as express or not. By integrating advanced algorithms, we can accurately assess the nature of new queries, ensuring more precise and timely classifications, even for previously unseen queries in our system.

Conclusion

This blog explored the design and implementation of Uber’s Presto express, aimed at reducing the end-to-end SLA for short-running queries. We discussed how express queries are defined, how they integrate into the overall ecosystem, and the challenges we encountered. Additionally, we highlighted the performance gains achieved through our deployment, with Presto express delivering an order-of-magnitude improvement in end-to-end SLA for over 75% of scheduled queries.