Presto_SQL_on_Everything.pdf_prestosql

presto

需积分: 0 104 浏览量更新于2023-05-24 评论收藏 543KB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源评论

资源推荐

Presto: SQL on Everything

Raghav Sethi, Martin Traverso

∗

, Dain Sundstrom

∗

, David Phillips

∗

, Wenlei Xie, Yutian Sun,

Nezih Yigitbasi, Haozhun Jin, Eric Hwang, Nileema Shingte

∗

, Christopher Berner

∗

Facebook, Inc.

Abstract—Presto is an open source distributed query engine

that supports much of the SQL analytics workload at Facebook.

Presto is designed to be adaptive, ﬂexible, and extensible. It

supports a wide variety of use cases with diverse characteristics.

These range from user-facing reporting applications with sub-

second latency requirements to multi-hour ETL jobs that aggre-

gate or join terabytes of data. Presto’s Connector API allows

plugins to provide a high performance I/O interface to dozens

of data sources, including Hadoop data warehouses, RDBMSs,

NoSQL systems, and stream processing systems. In this paper, we

outline a selection of use cases that Presto supports at Facebook.

We then describe its architecture and implementation, and call

out features and performance optimizations that enable it to

support these use cases. Finally, we present performance results

that demonstrate the impact of our main design decisions.

Index Terms—SQL, query engine, big data, data warehouse

I. INTRODUCTION

The ability to quickly and easily extract insights from large

amounts of data is increasingly important to technology-

enabled organizations. As it becomes cheaper to collect and

store vast amounts of data, it is important that tools to query

this data become faster, easier to use, and more ﬂexible. Using

a popular query language like SQL can make data analytics

accessible to more people within an organization. However,

ease-of-use is compromised when organizations are forced

to deploy multiple incompatible SQL-like systems to solve

different classes of analytics problems.

Presto is an open-source distributed SQL query engine that

has run in production at Facebook since 2013 and is used today

by several large companies, including Uber, Netﬂix, Airbnb,

Bloomberg, and LinkedIn. Organizations such as Qubole,

Treasure Data, and Starburst Data have commercial offerings

based on Presto. The Amazon Athena

interactive querying

service is built on Presto. With over a hundred contributors

on GitHub, Presto has a strong open source community.

Presto is designed to be adaptive, ﬂexible, and extensible.

It provides an ANSI SQL interface to query data stored in

Hadoop environments, open-source and proprietary RDBMSs,

NoSQL systems, and stream processing systems such as

Kafka. A ‘Generic RPC’

connector makes adding a SQL

interface to proprietary systems as easy as implementing a

half dozen RPC endpoints. Presto exposes an open HTTP

API, ships with JDBC support, and is compatible with sev-

eral industry-standard business intelligence (BI) and query

∗

Author was afﬁliated with Facebook, Inc. during the contribution period.

https://aws.amazon.com/athena

Using Thrift, an interface deﬁnition language and RPC protocol used for

deﬁning and creating services in multiple languages.

authoring tools. The built-in Hive connector can natively read

from and write to distributed ﬁle systems such as HDFS and

Amazon S3; and supports several popular open-source ﬁle

formats including ORC, Parquet, and Avro.

As of late 2018, Presto is responsible for supporting much

of the SQL analytic workload at Facebook, including interac-

tive/BI queries and long-running batch extract-transform-load

(ETL) jobs. In addition, Presto powers several end-user facing

analytics tools, serves high performance dashboards, provides

a SQL interface to multiple internal NoSQL systems, and

supports Facebook’s A/B testing infrastructure. In aggregate,

Presto processes hundreds of petabytes of data and quadrillions

of rows per day at Facebook.

Presto has several notable characteristics:

• It is an adaptive multi-tenant system capable of concur-

rently running hundreds of memory, I/O, and CPU-intensive

queries, and scaling to thousands of worker nodes while

efﬁciently utilizing cluster resources.

• Its extensible, federated design allows administrators to

set up clusters that can process data from many different

data sources even within a single query. This reduces the

complexity of integrating multiple systems.

• It is ﬂexible, and can be conﬁgured to support a vast variety

of use cases with very different constraints and performance

characteristics.

• It is built for high performance, with several key related

features and optimizations, including code-generation. Mul-

tiple running queries share a single long-lived Java Virtual

Machine (JVM) process on worker nodes, which reduces

response time, but requires integrated scheduling, resource

management and isolation.

The primary contribution of this paper is to describe the design

of the Presto engine, discussing the speciﬁc optimizations and

trade-offs required to achieve the characteristics we described

above. The secondary contributions are performance results for

some key design decisions and optimizations, and a description

of lessons learned while developing and maintaining Presto.

Presto was originally developed to enable interactive query-

ing over the Facebook data warehouse. It evolved over time to

support several different use cases, a few of which we describe

in Section II. Rather than studying this evolution, we describe

both the engine and use cases as they exist today, and call

out main features and functionality as they relate to these use

cases. The rest of the paper is structured as follows. In Section

III, we provide an architectural overview, and then dive into

system design in Section IV. We then describe some important

performance optimizations in Section V, present performance

results in Section VI, and engineering lessons we learned

while developing Presto in Section VII. Finally, we outline

key related work in Section VIII, and conclude in Section

IX. Presto is under active development, and signiﬁcant new

functionality is added frequently. In this paper, we describe

Presto as of version 0.211, released in September 2018.

II. USE CASES

At Facebook, we operate numerous Presto clusters (with sizes

up to ∼1000 nodes) and support several different use cases.

In this section we select four diverse use cases with large

deployments and describe their requirements.

A. Interactive Analytics

Facebook operates a massive multi-tenant data warehouse

as an internal service, where several business functions and

organizational units share a smaller set of managed clusters.

Data is stored in a distributed ﬁlesystem and metadata is stored

in a separate service. These systems have APIs similar to that

of HDFS and the Hive metastore service, respectively. We refer

to this as the ‘Facebook data warehouse’, and use a variant of

the Presto ‘Hive’ connector to read from and write to it.

Facebook engineers and data scientists routinely examine

small amounts of data (∼50GB-3TB compressed), test hy-

potheses, and build visualizations or dashboards. Users often

rely on query authoring tools, BI tools, or Jupyter notebooks.

Individual clusters are required to support 50-100 concurrent

running queries with diverse query shapes, and return results

within seconds or minutes. Users are highly sensitive to end-

to-end wall clock time, and may not have a good intuition

of query resource requirements. While performing exploratory

analysis, users may not require that the entire result set be

returned. Queries are often canceled after initial results are

returned, or use LIMIT clauses to restrict the amount of result

data the system should produce.

B. Batch ETL

The data warehouse we described above is populated with

fresh data at regular intervals using ETL queries. Queries are

scheduled by a workﬂow management system that determines

dependencies between tasks and schedules them accordingly.

Presto supports users migrating from legacy batch processing

systems, and ETL queries now make up a large fraction of

the Presto workload at Facebook by CPU. These queries

are typically written and optimized by data engineers. They

tend to be much more resource intensive than queries in the

Interactive Analytics use case, and often involve performing

CPU-heavy transformations and memory-intensive (multiple

TBs of distributed memory) aggregations or joins with other

large tables. Query latency is somewhat less important than

resource efﬁciency and overall cluster throughput.

C. A/B Testing

A/B testing is used at Facebook to evaluate the impact of

product changes through statistical hypothesis testing. Much of

the A/B test infrastructure at Facebook is built on Presto. Users

expect test results be available in hours (rather than days) and

that the data be complete and accurate. It is also important for

users to be able to perform arbitrary slice and dice on their

results at interactive latency (∼5-30s) to gain deeper insights.

It is difﬁcult to satisfy this requirement by pre-aggregating

data, so results must be computed on the ﬂy. Producing results

requires joining multiple large data sets, which include user,

device, test, and event attributes. Query shapes are restricted

to a small set since queries are programmatically generated.

D. Developer/Advertiser Analytics

Several custom reporting tools for external developers and

advertisers are built on Presto. One example deployment of

this use case is Facebook Analytics

, which offers advanced

analytics tools to developers that build applications which use

the Facebook platform. These deployments typically expose

a web interface that can generate a restricted set of query

shapes. Data volumes are large in aggregate, but queries

are highly selective, as users can only access data for their

own applications or ads. Most query shapes contain joins,

aggregations or window functions. Data ingestion latency is

in the order of minutes. There are very strict query latency

requirements (∼50ms-5s) as the tooling is meant to be inter-

active. Clusters must have 99.999% availability and support

hundreds of concurrent queries given the volume of users.

III. ARCHITECTURE OVERVIEW

A Presto cluster consists of a single coordinator node and

one or more worker nodes. The coordinator is responsible

for admitting, parsing, planning and optimizing queries as

well as query orchestration. Worker nodes are responsible for

query processing. Figure 1 shows a simpliﬁed view of Presto

architecture.

Worker

Data Source API

Processor

Worker

Coordinator

Planner/Optimizer

Scheduler

Metadata API

Data Location API

Queue

Processor

Query

Results

Data Source API

Processor

Worker

External

Storage

System

Fig. 1. Presto Architecture

The client sends an HTTP request containing a SQL state-

ment to the coordinator. The coordinator processes the request

https://analytics.facebook.com

by evaluating queue policies, parsing and analyzing the SQL

text, creating and optimizing distributed execution plan.

The coordinator distributes this plan to workers, starts exe-

cution of tasks and then begins to enumerate splits, which are

opaque handles to an addressable chunk of data in an external

storage system. Splits are assigned to the tasks responsible for

reading this data.

Worker nodes running these tasks process these splits by

fetching data from external systems, or process intermediate

results produced by other workers. Workers use co-operative

multi-tasking to process tasks from many queries concurrently.

Execution is pipelined as much as possible, and data ﬂows

between tasks as it becomes available. For certain query

shapes, Presto is capable of returning results before all the

data is processed. Intermediate data and state is stored in-

memory whenever possible. When shufﬂing data between

nodes, buffering is tuned for minimal latency.

Presto is designed to be extensible; and provides a versa-

tile plugin interface. Plugins can provide custom data types,

functions, access control implementations, event consumers,

queuing policies, and conﬁguration properties. More impor-

tantly, plugins also provide connectors, which enable Presto to

communicate with external data stores through the Connector

API, which is composed of four parts: the Metadata API, Data

Location API, Data Source API, and Data Sink API. These

APIs are designed to allow performant implementations of

connectors within the environment of a physically distributed

execution engine. Developers have contributed over a dozen

connectors to the main Presto repository, and we are aware of

several proprietary connectors.

IV. SYSTEM DESIGN

In this section we describe some of the key design decisions

and features of the Presto engine. We describe the SQL dialect

that Presto supports, then follow the query lifecycle all the way

from client to distributed execution. We also describe some

of the resource management mechanisms that enable multi-

tenancy in Presto. Finally, we brieﬂy discuss fault tolerance.

A. SQL Dialect

Presto closely follows the ANSI SQL speciﬁcation [2]. While

the engine does not implement every feature described, im-

plemented features conform to the speciﬁcation as far as

possible. We have made a few carefully chosen extensions to

the language to improve usability. For example, it is difﬁcult

to operate on complex data types, such as maps and arrays,

in ANSI SQL. To simplify operating on these common data

types, Presto syntax supports anonymous functions (lambda

expressions) and built-in higher-order functions (e.g., trans-

form, ﬁlter, reduce).

B. Client Interfaces, Parsing, and Planning

1) Client Interfaces: The Presto coordinator primarily ex-

poses a RESTful HTTP interface to clients, and ships with

a ﬁrst-class command line interface. Presto also ships with a

JDBC client, which enables compatibility with a wide variety

of BI tools, including Tableau and Microstrategy.

2) Parsing: Presto uses an ANTLR-based parser to convert

SQL statements into a syntax tree. The analyzer uses this

tree to determine types and coercions, resolve functions and

scopes, and extracts logical components, such as subqueries,

aggregations, and window functions.

3) Logical Planning: The logical planner uses the syntax

tree and analysis information to generate an intermediate

representation (IR) encoded in the form of a tree of plan nodes.

Each node represents a physical or logical operation, and the

children of a plan node are its inputs. The planner produces

nodes that are purely logical, i.e. they do not contain any

information about how the plan should be executed. Consider

a simple query:

SELECT

orders.orderkey, SUM(tax)

FROM orders

LEFT JOIN lineitem

ON orders.orderkey = lineitem.orderkey

WHERE discount = 0

GROUP BY orders.orderkey

The logical plan for this query is outlined in Figure 2.

Aggregate [SUM(tax)]

LeftJoin [ON orderkey]

Scan [orders]

Filter [discount=0]

Scan [lineitem]

Fig. 2. Logical Plan

C. Query Optimization

The plan optimizer transforms the logical plan into a more

physical structure that represents an efﬁcient execution strategy

for the query. The process works by evaluating a set of

transformation rules greedily until a ﬁxed point is reached.

Each rule has a pattern that can match a sub-tree of the

query plan and determines whether the transformation should

be applied. The result is a logically equivalent sub-plan that

replaces the target of the match. Presto contains several rules,

including well-known optimizations such as predicate and

limit pushdown, column pruning, and decorrelation.

We are in the process of enhancing the optimizer to perform

a more comprehensive exploration of the search space using

a cost-based evaluation of plans based on the techniques

introduced by the Cascades framework [13]. However, Presto

already supports two cost-based optimizations that take table

and column statistics into account - join strategy selection and

join re-ordering. We will discuss only a few features of the

optimizer; a detailed treatment is out of the scope of this paper.

1) Data Layouts: The optimizer can take advantage of

the physical layout of the data when it is provided by the

connector Data Layout API. Connectors report locations and

other data properties such as partitioning, sorting, grouping,

剩余11页未读，继续阅读

xia0cai

粉丝: 9
资源: 1

会员权益专享

Presto_SQL_on_Everything.pdf

评论0

会员权益专享

最新资源

Presto_SQL_on_Everything.pdf

评论0

Presto最新版集成到Ambari

如何解决Could not initialize class io.prestosql.teradata.functions.TeradataDateFunctions

如何使用大数据presto集成开发,转换分到元？

- 对比druid，clickhouse,kylin,kawq,presto，spark sql，impala等主流开源离线数据分析引擎

druid/impala/presto/spark sql/kylin/elasticsearch功能对比

presto on spark：扩展 presto 以支持大规模 etl

presto 日期函数

presto、druid、sparksql、kylin的对比分析

clickhouse和presto对比

presto大数据中报错，line 50:5: mismatched input 'and'. Expecting:，可能原因为

哪些数据源是通过Starburst Presto的扩展连接提供的

使用presto大数据比较timestamp时间大小

如何比较hive，spark，impala和presto？

presto的基础语法

presto与hive的区别

presto底层是怎么计算的，用的什么计算引擎

presto引擎如何查询两个表,返回两个表中不一样的值

在github上关于Doris的项目有哪些？

presto和impala有什么差异

会员权益专享

最新资源