论文链接：https://trino.io/Presto_SQL_on_Everything.pdf

摘要

Presto 是一个开源的分布式查询引擎，Facebook 内部大部分的 SQL 分析工作均由其支持。Presto 设计上兼顾了适应性、灵活性和可扩展性。其支持了很多不同特征的用例。这些用例的执行时间从亚秒级用户报表到小时级的 ETL 作业（agg 或 join 等复杂算子，TB 数据量）。Presto 的连接器 API 以插件的方式适配了数十个数据源，包括 Hadoop、RDBMS、NoSQL 和流式处理系统。本文，首先概述了 Presto 在 Facebook 内部支撑的多个经典用例。然后，描述了 Presto 的架构及其实现，并阐述了以更好的支持这些用例的功能点和性能优化项。最后，展示了主要设计决策对性能结果的影响。

I. 引言

掌握从大量数据中快速轻松地分析洞察的能力对技术型组织来说越来越重要。由于收集和存储大量数据的成本越来越低，导致能够快速、易用、灵活地查询这些数据的工具变得更加重要。通用的 SQL 交互方式会使组织内的更多人可以进行数据分析。然而，当组织被迫部署多个不兼容的类 SQL 的系统时，反而易用性就会变差。

Presto 是一种开源分布式 SQL 查询引擎，2013 年以来一直在 Facebook 的生产环境中运行，并且现在已被多个大公司使用，例如：Uber、Netflix、Airbnb、Bloomberg 和 LinkedIn。Qubole、Treasure Data 和 Starburst Data 等都有基于 Presto 的商业产品。Amazon Athena 是基于 Presto 构建的一种交互式查询服务。Presto 开源社区强大，拥有一百多名贡献者。

适应性、灵活性和可扩展性是 Presto 的设计目标。其提供了一个 ANSI 标准 SQL 接口来查询多种存储数据，比如：Hadoop环境、开源和专有的 RDBMS 数据库、NoSQL 系统和Kafka等流处理系统。“通用 RPC”连接器使得向专有系统添加 SQL 接口就像实现几个 RPC 接口一样简单。Presto 暴露了一个开放的 HTTP API，支持了JDBC，并兼容了多个行业标准的商业智能（Business Intelligence, BI）和查询编写工具。内置的 Hive 连接器可以“本地”亲和地读取和写入分布式文件系统，例如 HDFS 和 Amazon S3；并支持多种流行的开源文件格式，包括 ORC、Parquet 和 Avro。

截至 2018 年底，Presto 支撑了 Facebook 中非常多的 SQL 分析工作，包括交互式分析、商业智能查询和长时间运行的批处理“提取-转换-加载 (ETL)”作业。此外，Presto 支持多个面向终端用户的分析工具，提供高性能仪表盘，为多个内部 NoSQL 系统提供了 SQL 接口，支撑了 Facebook 的 AB 测试平台。在 Facebook 中，Presto 每天处理数百 PB 的数据量，行数约为千万亿。

Presto 有几个显著特性：

• 它是一个自适应的多租户系统。能同时运行数百个内存、I/O 和 CPU 密集型查询，即使集群 Worker 节点扩展到数千个，资源利用率依旧非常好。
• 它具有较好的扩展性。在同一个（联邦）查询中可以处理来自多个不同数据源的数据，这降低了集成多个系统的复杂度。
• 它具有较好的灵活性。不同限制条件及性能特征各异的多个用例可以共用同一份集群配置。
• 它是为高性能而构建的，具有几个关键的相关特征和优化项，比如代码生成。Worker 节点上多个同时运行的查询可以同时使用同一个长时间运行的 JVM，实现了任务调度、资源管理和隔离等功能，减少了响应时间。

本论文主要描述了 Presto 引擎的设计方案，讨论了实现上述特性所需的特定优化及其权衡，提供了一些关键设计决策和优化项的性能结果，总结了开发和运维 Presto 的经验教训。

Presto 最初开发是为了实现对 Facebook 数仓的交互式查询。随着时间推移演进，Presto 已可以支持多种不同的用例，第 II 节会描述其中的一些用例。本文没有研究这种演进，而是描述了 Presto 引擎和当前已经存在和用例，并说明了这些用例的主要特征及其功能。第 III 节提供了架构概述，第 IV 节深入研究了系统设计。第 V 节描述了一些重要的性能优化策略，第 VI 节展示了性能结果，第 VII 节总结了开发 Presto 的工程经验。最后，第 VIII 节概述了关键的相关工作，第 IX 节是总结。Presto 正在积极开发，重要的新功能也会被添加进来。本论文描述的 Presto 是 2018 年 9 月发布的 0.211 版本。

II. 用例

在 Facebook，我们运维着大量的支持不同用例的 Presto 集群（最多 1000 个节点）。本小节选取了四个被大规模部署的用例，并描述了这些用例的需求。

A. 交互式分析

Facebook 运营着一个大规模的多租户数据仓库作为内部服务，其中多个业务和部门共享一组较小的托管集群。数据存储在分布式文件系统，元数据存储在其它服务。这些系统均提供了类似 HDFS 存储和 Hive 元数据服务的 API。我们将其称为“Facebook 数据仓库”，采用类似 Presto Hive 的连接器对其进行读取和写入。

Facebook的工程师和数据科学家通常借助查询编写工具、BI 工具或 Jupyter 笔记本等客户端，进行针对少量数据（压缩后约 50GB~3TB）定期检查、验证假设、构建可视化组件或仪表盘等操作。单个集群需要支持 50-100 个不同格式的并发查询，并需要在几秒或几分钟内返回结果。用户对端到端处理时间非常敏感，并且可能不清楚查询需要多少硬件资源。当执行探索性分析时，用户可能不需要返回完整的结果集。返回初始结果后，查询通常会被取消，或者会使用 LIMIT 条款来限制系统应产生的结果数量。

B. 批处理 ETL

数据仓库通常使用 ETL 查询定期刷新数据。工作流管理系统根据任务间的依赖关系调度多个查询。在 FaceBook 中，Presto 中的很多 ETL 是从其他批量处理系统迁移过来的，并且这部分 ETL 作业占用了大量的 CPU 负载。这些查询通常由数据工程师编写和优化。与交互式分析相比，由于会执行 CPU 密集型转换和内存密集型（TB 级的分布式内存）的 agg 和多表 join 等复杂操作，因此这部分查询通常会占用更多的硬件资源。跟资源效率和整个集群吞的吐量相比，查询延时反而显得不太重要了。

C. AB 测试

Facebook 使用 AB 测试平台通过统计假设检验来评估产品变更的影响。Facebook 的大部分 AB 测试平台都是基于 Presto 构建的。用户希望在数小时（而不是数天）内得到测试结果，并且结果数据完整准确。用户能够在交互式延迟（ 5-30 秒）内对其结果执行任意切片和切块，以获得更深入的见解，也非常重要。通过预聚合数据很难满足这一要求，因此结果必须被快速计算。执行过程中需要 join 多个大型数据集，其中包括用户、设备、测试和事件属性。由于这类查询是以编程方式生成的，因此查询样式较少。

D. 开发者/广告商分析

基于 Presto 构建了多种面向外部开发者和广告商的自定义报表工具。其中一个示例部署是 Facebook Analytics，其为使用 Facebook 平台构建应用程序的开发者提供了高级分析工具。这些部署通常会暴露一个可以生成一组受限查询样式的 Web 界面。聚合的数据量很大，但查询却具有高度的选择性（过滤性），因为用户只能访问自己的应用程序或广告的数据。大多数查询样式包含 join、agg 或 window 函数。数据采集延迟是分钟级的。由于是交互式的，因此有非常严格的查询延迟要求（50ms~5s）。集群必须具有 99.999% 的可用性，并在给定用户量的情况下能够支持数百个并发查询。

III. 架构概览

Presto 集群由单个 Coordinator 节点和一个或多个 Worker 节点组成。Coordinator 负责接收、解析、生成和优化查询以及查询调度。Worker 节点负责查询处理。图 1 展示了 Presto 架构的简化图。

客户端向 Coordinator 发送一个包含 SQL 语句的 HTTP 请求。Coordinator 通过评估队列策略、解析和分析 SQL 文本、创建和优化分布式执行计划来处理请求。

Coordinator 将此计划分发给 Worker 节点（多个），开始执行任务，然后开始枚举分片，这些分片是外部存储系统中可寻址数据块的不透明句柄。分片被分配给负责读取此数据的任务。

Worker 节点从外部系统拉取这些分片对应的数据，或者处理其他 Worker 节点产生的中间结果。Worker 节点可以并行处理来自多个查询的多个任务。采用管道式流式执行，数据尽可能的在多个任务间流动。对于某些特定的查询样式， Presto 可以在处理完所有数据之前就返回结果。中间数据和状态尽可能存储在内存。采用缓冲区以降低节点间交互的延时。

Presto 被设计为可扩展的，并提供了一个通用的插件接口。插件可以提供自定义数据类型、函数、访问控制实现、事件消费者、排队策略和配置属性。更重要的是，插件还可以提供“连接器”，使 Presto 能够通过连接器 API 与外部数据存储进行通信。连接器 API 由四部分组成：元数据 API、数据位置 API、数据源 API 和数据接收器 API。这些 API 旨在允许在分布式执行引擎的物理环境中实现高性能的连接器。开发人员已经为 Presto 主分支代码库贡献了十多个连接器，其中包含了多个大家熟知的专有连接器。

IV. 系统设计

本节将介绍 Presto 引擎的一些关键设计决策和功能。描述了 Presto 支持的 SQL 方言，然后是从客户端到分布式执行的整个查询生命周期。还描述了在 Presto 中启用多租户的一些资源管理机制。最后，简要讨论了容错。

A. SQL 方言

Presto 严格（closely）遵循 ANSI SQL 规范。虽然引擎不会实现规范描述的每个功能，但已实现的功能尽可能符合规范。 而且，对该语言进行了一些精心选择的扩展以提高可用性。比如，在 ANSI SQL 中很难对复杂数据类型（如 Map 和 Array）进行操作。 为了简化对这些通用数据类型的操作，Presto 语法支持匿名函数（lambda 表达式），且内置了很多高优先级的 transform、filter 和 reduce 等函数。

B. 客户端接口、解析和计划

1. 客户端接口

Presto Coordinator 向客户端暴露了 RESTful HTTP 接口，并附带了较好的命令行接口。Presto 还附带了一个 JDBC 客户端，依此 Presto 可以兼容包括 Tableau 和 Microstrategy在内的多个 BI 工具。

2. 解析

Presto 使用基于 ANTLR 构建的解析器将 SQL 语句转换为语法树（AST，Abstract Syntax Tree）。分析器使用此语法树来确定数据类型和强制类型转换，解析函数及作用域，并提取 sub-query、agg 和 window 函数等逻辑组件。

3. 逻辑计划

使用语法树和分析信息，逻辑计划器将生成以计划节点树编码的中间表示（IR）。每个节点代表一个物理或逻辑操作，计划节点的子节点是其输入。计划器生成纯逻辑的节点，不包含计划执行的任何信息。

图2 展示了上面简单查询对应的逻辑计划。

C. 查询优化

计划优化器将逻辑计划转换为进一步能够高效执行的物理结构。此处理过程会启发式循环执行一组转换规则，直到达到稳定为止。每个规则有一个匹配查询计划的一个子树的模式，依此可以判定是否需要执行此规则对应的转换。采用转换之后的子计划代替匹配子树。Presto 内置了多个优化规则，其中包含了谓词下推、limit下推、列裁剪和关联性去除等大家熟知的规则。

Presto 正在使用基于 Cascades 框架的增强优化器，其采用 CBO 技术来更加全面地探索查询空间。目前，Presto 已支持“join 策略选择”和“join 重排序”两种 CBO 优化规则，这两个规则均参考了表和列的统计信息。本文只讨论优化器的几个简要功能。

1. 数据布局

当连接器数据布局 API 可以提供物理布局的数据时，优化器利用此数据执行效果更佳。连接器提供了位置信息和其他数据属性，比如：分区、排序、分组和索引。连接器可以为单个表提供多个具有不同属性的布局，优化器可以针对查询选择最有效的布局。在开发者/广告商分析系统，此功能通常由可以操作集群的管理员使用；管理员可以简单地添加物理布局来优化新的查询样式。后续小节将进一步描述引擎利用这些属性的一些策略。

2. 谓词下推

优化器可以与连接器一同决定是否可以下推范围（range）和相等谓词条件，通过连接器优化过滤效率。

例如，开发者/广告商分析用例采用构建在分片的 MySQL 之上的专有连接器。连接器将存储在单个 MySQL 实例中的数据划分成多个分片，并且可以将范围或相等谓词下推到各个分片，依此确保只从 MySQL 读取匹配的数据。如果存在多个布局，引擎会选择在谓词列上建立了索引的布局。在开发者/广告商分析用例中，基于索引的过滤是一种高选择性的过滤器。对于交互式分析和批处理 ETL 用例，Presto 利用 Hive 连接器中的分区裁剪和文件格式功能也以类似的方式提高了性能。

3. 节点间并行性

优化过程的一个步骤是识别可以跨 Worker 节点并行执行的计划部分。这些部分被称为“阶段（stage）”，每个阶段将对应一个或多个任务，每个任务对不同的输入数据集执行相同的计算。引擎在阶段之间插入可以内存缓冲的数据传输节点（RemoteExchangeNode）以实现数据交换（shuffle）。 shuffle 会增加延迟、耗尽缓冲内存并具有高 CPU 开销。因此，优化器必须仔细推理计划中引入的 shuffle 总数。图 3 展示了如何将计划划分为多个阶段并使用 shuffle 将它们连接起来的一个简单实现。

数据布局属性

优化器可以利用物理数据布局最小化计划中的 shuffle 数。这在 AB 测试用例中非常有用，因为几乎每个查询都需要一个大 join 来生成实验明细或人口信息。该引擎利用参与 join 的两个表都在同一列上分区的事实，并使用基于位置协作的 join 策略来消除资源密集型 shuffle。

如果连接器提供一个连接列被标记为索引的数据布局，优化器能够确定使用索引嵌套循环连接是否将是一个最佳策略。可以通过 join 线上数据存储（键值或其他）来非常有效地操作存储在数据仓库中的规范化数据。这是交互式分析用例中的常用功能。

节点属性

跟连接器类似，计划树中的节点也可以表达其输出属性（数据的分区、排序、分桶和分组等特征）。这些节点还能够表达必需的（required）和偏好的（preferred）属性，在引入 shuffle 时会参考这些属性。冗余 shuffle 被简单地删除了，但在其他情况下，可以更改 shuffle 的属性以减少所需的 shuffle 次数。Presto 会贪婪地选择满足尽可能多的所需属性的分区，以减少shuffle。这意味着优化器可能会选择在更少的列上进行分区，但这在某些情况下会导致更大的分区倾斜。例如，图 3 中的计划应用此优化后，会折叠为单个数据处理阶段。

4. 节点内并行

优化器使用类似的机制来识别计划阶段内的部分，这些部分可以从单节点多线程并行化中受益。节点内并行化比节点间并行化更有效，因为延迟开销很小，并且状态（比如：哈希表和字典）可以在线程之间被有效共享。提升节点内并行性可以显著加快速度，特别是对于并发性限制下游阶段吞吐量的查询样式：

• 交互式分析涉及运行许多短的一次性查询，用户通常不会花时间尝试优化这些查询。 因此，分区倾斜很常见，要么是由于数据的固有属性，要么是由于常见的查询模式（例如，按用户国家/地区分组，同时也过滤到一小部分国家/地区）。这通常表现为大量数据被哈希分区到了少量节点上。
• 批处理 ETL 作业通常在很少或没有过滤的情况下直接转换大型数据集。在这些场景中，计划树的较高层级涉及的节点数量较少，可能不足以快速处理叶子阶段生成的数据量。

这两类场景，各个 Worker 节点均采用多个线程执行计算可以在一定程度上减轻这种并发瓶颈。引擎可以采用多线程模式运行单个算子序列（管道）。图 4 展示了优化器如何并行化 join 的一个部分。

D. 调度

Coordinator 以可执行任务的形式将计划阶段分布式调度到 Worker 节点，可以将执行任务看作单个处理单元。然后，Coordinator 将一个阶段的任务链接到其他阶段的任务，形成一个通过 shuffle 相互链接的处理器树。数据尽可能的从一个阶段流到另一个阶段。

一个任务可能包含多个管道。管道由一系列算子组成，每个算子对数据执行单一的明确定义的计算操作。例如，一个执行哈希 join 的任务必须包含至少两条管道，一个构建哈希表（构建管道），另一个从探测端流式获取数据并执行 join（探测管道）。当优化器确定管道的一部分可以在本地（ Worker 节点）并行执行时，可以拆分此管道并独立地并行化该部分。图 4 展示了将“构建管道”如何拆分为两个管道，一个用于扫描数据，另一个用于构建哈希表的分区。管道通过本地内存 shuffle 关联在一起。

为了执行查询，引擎制定了两类调度策略。第一类用于确定阶段的调度顺序，第二类用于确定应该调度为多少个任务，以及这些任务应该放置在哪些节点上。

1. 阶段调度

Presto 支持两种阶段调度策略：整体调度策略（all-at-once）和分阶段调度策略（phased）。整体调度策略通过同时调度所有阶段，数据一旦可用就会被处理，各个阶段可以并行执行，依此尽量减少挂钟时间（wall clock time）。这种调度策略适应于延迟敏感的用例，例如交互式分析、开发者/广告商分析和 AB 测试。分阶段调度策略必须同时启动有向数据流图的所有强连接组件，以避免死锁，并按拓扑顺序执行这些组件。例如，如果以分阶段调度策略执行哈希 join，则在构建完散列表之前不能调度 join 左侧流的任务。这大大提高了批处理分析用例的内存效率。

当调度程序确定应根据策略调度某个阶段时，其还需要将此阶段的任务分配给 Worker 节点。

2. 任务调度

任务调度器检查计划树并将阶段分为叶子阶段和中间阶段两类。叶子阶段从连接器读取数据，而中间阶段只处理来自其他阶段的中间结果。

叶子阶段

叶子阶段对应的任务调度器会同时兼顾网络因素和连接器本身的限制条件。比如：无共享部署要求 Worker 节点和存储节点需在同一机器上。这种情况下，任务调度器会根据连接器的数据层 API 决定任务分配的节点。AB 测试用例需要可预期的高吞吐率和低延时的数据读取机制，而 Raptor 连接器可以满足这些要求。Raptor 是一个针对 Presto 优化的无共享模式的存储引擎，其将 ORC 文件存放于闪存，元数据则存放于 MySQL。

分析显示，生产集群中的大部分 CPU 时间花费在了针对连接器读取数据的处理上，这些热点处理包含：解压缩、解码、过滤和应用转换。这项工作具有较高的并行度，可以在尽可能多的节点上运行这类阶段，以取得最短的挂钟时间。如果没有限制，这些数据则可以划分为足够多的分片，并且叶子阶段对应的任务可以调度到任意 Worker 节点。 Facebook 数据仓库部署为可远程访问的共享存储模式，集群中的任意节点都可以参与处理叶子节点。但这种执行策略可能是网络密集型的。

调度器还可以推理网络拓扑，以利用插件提供的层次去优化数据读取。Facebook 部署的受限网络结构可以利用这种机制向引擎表达“机架本地读取策略”优先于“机架远程读取策略”。

中间阶段

中间阶段的任务可以分配到任何 Worker 节点。然而，引擎仍需要决定每个阶段应该被调度为多少个任务。任务数取决于连接器配置、计划本身的属性、所需的数据布局和其他的部署配置。在某些情况下，引擎可以在执行期间动态改变此任务数。

3. 分片调度

当叶子阶段对应的任务在 Worker 节点上执行时，此 Worker 节点可以接受一个或多个分片。不同连接器对应的分片信息也会不同。分布式文件系统对应的分片信息可能包含文件路径和文件区域的偏移量。Redis 键值存储对应的分片信息则会包含表信息、键值格式和要查询的主机列表等等。

叶子节点对应的各个任务只有被分配一个或多个分片才有资格运行。而中间阶段对应的任务总是有资格运行的，只有自身被挂起或者其上游任务全部完成之后才能结束。

分片分配

随着任务在 Worker 节点上启动，Coordinator 开始分配分片给这些任务。Presto 要求连接器可以枚举小批量分片，并将这些分片延时分配给任务。这是 Presto 的一个重要功能，依此带来了如下几个好处：

• 将查询响应的时间和连接器枚举大量分片的时间相互解耦。例如，Hive 连接器需要花费几分钟的时间才能完成枚举分区及每个分区目录的文件列表。
• 查询无需处理完所有数据就可以开始生成结果，当遇到 LIMIT 条款时，查询通常可以快速取消或者提前结束。在交互式分析用例中，很多查询在枚举完所有分片之前就已经结束。
• Worker 节点维护了一个已分配的待处理的分片队列。Coordinator 会将新分片分配给最短队列对应的节点。保持这些队列较小，依此系统能够自动适配分片之间 CPU 处理成本不同和 Worker 节点之间的性能差异。
• 允许查询执行时无需将所有元信息全量加载到内存。由于查询可能访问数百万个分片，导致 Coordinator 所有可用内存会被轻松消耗掉，因此这对于 Hive 连接器是重要的。

对于兼容 Hive 的 Facebook 数据仓库，这些特征对于交互式分析和批处理 ETL 用例特别有用。值得注意的是，延时分片枚举机制同时会导致很难准确地评估和报告查询进度。

E. 查询执行

1. 本地数据流

一旦分片被分配给线程，其就会由驱动（driver）循环执行。Presto 驱动循环比递归迭代器中流行的 Volcano（拉）模型更复杂，但提供的功能是重要的。其更适合协作式多任务处理，因为算子可以在让出线程之前将快速进入已知状态，而不是无限期地阻塞。此外，驱动可以通过在无需额外输入（例如，恢复资源密集型或爆炸性转换的计算）即可实现算子间移动数据，依此可最大限度地提升单位调度时间内执行的工作。循环的每次迭代都会在所有能够取得进展的算子间移动数据。

驱动循环操作的数据单元成为页（page），其由列式编码的一系列行组成。当传入分片时，连接器数据源 API 会返回页。算子通常消费输入页，执行计算，然后生成输出页。图 5 展示了页的内存结构。驱动循环会持续地移动算子间的页，直到完成单位调度时间，或者直到算子不能取得进展。

2. Shuffles

Presto 旨在最大限度地减少端到端延迟，同时最大限度地提高资源利用率，节点间数据流机制反映了这一设计选择。Presto 采用基于内存缓冲区的 HTTP 协议实现了中间结果交互。任务生产的数据存储于缓冲区以供其他Worker 节点消费。Worker 节点采用 HTTP 长轮询模式从其他 Worker 节点请求中间结果。服务器将会保留数据，直到客户端使用前一个响应发送的令牌请求下一段数据。这也意味着，在传输协议中隐含了确认机制。这种长轮询机制缩短了响应时间，尤其是在传输少量数据时。与其他将 shuffle 数据落盘的系统相比，这种机制提供的延时低很多，依此 Presto 可以支持开发者/广告商分析等延时敏感的用例。

引擎调节并行度以维持输出和输入缓冲区的利用率。填满的输出缓冲区会导致分片执行停止和耗尽宝贵的内存，而未充分利用的输入缓冲区则会增加不必要的处理开销。

引擎会持续地监控输出缓冲区的利用率。当利用率持续很高时，其会通过减少可执行的分片数来降低有效的并发度。这会提升共享网络资源的公平性。在客户端（最终用户或者其他 Worker 节点）不能及时消费已生成数据时，这也是一项重要的效率优化。如果没有此功能，运行复杂多阶段查询可能会长时间持有数十 GB 的缓冲区内存。即使在交互式分析用例中，采用 BI 或 查询编写工具通过慢速连接下载少量结果数据（10~50MB）时，这种情况也经常发生。

在接收方，引擎会监控每个请求的平均传输数据量，依此计算目标 HTTP 请求并行度，在不超过容量的基础上尽量使输入缓冲区保持填充状态。这种背压导致上游任务在缓冲区填满时会变慢。

3. 写

ETL 作业通常生成必须写入其他表的数据。影响远程存储环境写性能的一个重要的驱动因素是执行写入的并行度（通过连接器数据接收器 API 写数据的线程的总数量）。

以 Hive 连接器使用 Amazon S3 作为存储为例。针对 S3 的每次并行写都会创建一个新文件，而少量聚合数据的数百次的写入可能会创建很多小文件。除非这些小数据单元能够被稍后合并，否则读取时可能产生不可接受的高开销（许多缓慢的元数据操作和延迟受限的读取性能）。然而，采用太小的并行度则可能会使总的写入吞吐降到不可接受的水平。Presto 再次采用了自适应方法，当引擎确定用于写入的生产数据阶段超出了缓冲区利用率阈值（配置的单次写入数据操作的写入阈值）的时候，采用在更多 Worker 节点添加任务的方式可以动态增加写入的并行度。对于写入密集型的批处理 ETL 用例来说，这是一个重要的效率优化。

F. 资源管理

使 Presto 非常适合多租户部署的其中一个关键特性是其包含了一个全量集成的细粒度资源管理系统。单个集群能够并行执行数百个查询。并可以最大限度地利用 CPU、IO 和内存资源。

1. CPU调度

Presto 要针对整体集群的吞吐进行优化，比如：汇总用于处理数据的总 CPU。本地调度器还优化了计算成本低的查询的周转时间，且要求相似的的查询之间可以公平的共享 CPU 资源。任务的资源使用情况由每个分片的线程 CPU 时间汇总得到。为了最大限度地减少协调开销，Presto 跟踪了任务级的 CPU 资源使用情况，并做出了本地调度决策。

Presto 采用协作多任务模型在每个 Worker 节点上调度了多个并发任务，实现了多租户功能。任意给定的分片只允许在一个线程上运行，且最大的单位调度时间设置为 1 秒，之后此任务必须释放线程并返回到队列。当输出缓冲区已满（下游阶段不能及时消费数据）、输入缓冲区为空（上游阶段不能及时生产数据）或者系统内存不足时，本地调度器甚至会在单位调度时间完成前简单地切换以处理另一个任务。这能够将线程释放给可执行的分片，有助于提升 Presto 的 CPU 利用率，并可以高度适应不同的查询样式。所有用例均会收益于这种精细的资源效率。

当分片释放一个线程时，引擎需要决定下一个要运行的任务（与一个或多个分片相关联）。Presto 不是提前预测完成新查询所需的资源，而是简单地使用任务的总 CPU 时间将其分类为五个级别的多级反馈队列。随着任务积累更多的 CPU 时间，它们会移动到更高的级别。每个级别都分配了一个可用 CPU 时间的配置比例。在实践中，针对任意工作负载，都能完成公平地协调多任务具有挑战性。分片（有时甚至在同一任务中）的 I/O 和 CPU 特性差异非常大，包含复杂函数（例如，正则表达式）的分片相对于其他分片可能消耗更多的线程时间。有一些连接器没有提供异步 API，导致工作线程可能被保持几分钟。

调度器在处理这些约束时必须是自适应的。该系统提供了一个低成本的释放信号量，因此可以在算子内停止长时间运行的计算操作。如果算子超过了单位调度时间，调度器会向任务“收取”实际线程时间，并暂时减少未来的执行次数。Presto 为资源消耗最低的查询提供了较高的优先级，这种自适应行为能够应对交互式分析和批处理 ETL 用例中查询样式的多样性。此决策反映了这样一种理解：用户期望计算成本低的查询能够快速完成，而不太关心大型、计算成本高的作业的周转时间。由于短查询会快速退出系统，因此同时运行非常多的查询，即使上下文切换代价较多，也会具有较短的总排队时间。

2. 内存管理

像 Presto 这样的多租户系统，内存是主要的资源管理挑战之一。在本节中，描述了引擎控制集群内存分配的机制。

内存池

Presto 中所有主要的内存分配都必须归类为用户内存或系统内存，并在相应的内存池中预留内存。用户内存是一种用户只根据系统或者输出数据的基本知识就可以推断出的内存使用情况（例如，聚合的内存使用与其基数成正比）。另一方面，系统内存通常是实现决策（例如，shuffle 缓冲区）的副产品的内存使用情况，可能与查询样式和输入数据量无关。

引擎针对用户内存和总（用户+系统）内存分别施加了单独的限制，超过全局限制（跨 Worker 节点的总和）或者单节点限制的查询将被终止。当节点内存不足时，会通过暂停任务处理的方式来阻塞查询的内存预留操作。总内存限制通常比用户限制设置的更高，生产环境只有少数查询会超过总内存限制。

查询的单节点和全局的总用户内存限制通常是不同的，这可以最大限度的允许倾斜情况的发生。考虑一个 500 节点的集群，单节点有 100GB 的可用查询内存，而单个查询的全局内存最多可以使用 5TB。这种情况下，可以同时分配 10 个查询以达到总内存量。但如果想要允许 2:1 的倾斜（查询的一个分区消耗是平均数的 2 倍），则单节点查询内存限制必须设置为 20GB。这意味着，在不耗尽可用节点内存的情况下，只保证可以同时运行 5 个查询。

能够在 500 个节点的交互式分析或批处理 ETL 集群上同时运行 5 个以上的查询是重要的。由于集群中给定的查询内存特征（倾斜、分配比例和分配时间的局部性）差异很大，因此在任何时间这 5 个查询在同一个 Worker 节点上分配的内存均不太可能达到单节点内存上限。由此，当节点内存不足时，只要存在保持集群健康的机制，过度使用集群的内存通常也是安全的。Presto 中有两种这样的健康机制：溢出和预留池。

溢出

当一个节点内存不足时，引擎会按执行时间的升序对符合条件的任务调用内存回收操作，直到有足够内存可满足最近一个请求时，才会停止上述的回收操作。回收是通过将状态溢出到磁盘来处理的。Presto 支持哈希 join 和 agg 的溢出。但Facebook 的部署环境均没有配置为可溢出。集群规模通常足以支持数 TB 的分布式内存，而且完全采用内存，执行的延时也是可预期的，因此用户更加欢迎。并且，本地磁盘也会增加硬件成本（尤其是在 Facebook 的共享存储部署中）。

预留池

如果一个节点内存不足，且集群没有配置为可溢出，或者没有剩余的可回收内存，则采用预留内存机制来解除集群阻塞。每个节点上的查询内存池进一步细分为了两个池：通用池和预留池。当某个 Worker 节点上的通用池耗尽时，此 Worker 节点上使用内存最多的查询将“提升”到所有 Worker 节点上的预留池。在这种状态下，分配给此查询的内存被计入预留池而不是通用池。为了防止死锁（不同 Worker 节点停止不同的查询），整个集群只有一个查询可以进入预留池。 如果一个节点上的通用池耗尽，而预留池被占用，则会停止该节点上其他任务的所有内存请求。直到预留池中的查询运行完成，集群才会将所有未完成的内存请求的阻塞解除。这种决策这有点浪费，因为每个节点上的预留池大小必须适配于查询运行的本地内存限制。通过配置，集群可以决定是否终止“阻塞大多数节点”的查询。

G. 容错

Presto 能够使用低级重试机制恢复许多的短暂错误。然而，截至 2018 年底，Presto 仍没有任何有意义的内置容错机制，来解决 Coordinator 或 Worker 节点崩溃故障问题。Coordinator 故障将导致集群变得不可用， Worker 节点崩溃故障则会导致该节点上运行的所有查询都会失败。Presto 依赖客户端自动重试失败的查询。

在 Facebook 的生产环境中，采用外部编排机制根据用例已不同的可用性模式运行了多个集群。交互式分析和批处理 ETL 用例运行了备用的 Coordinator，而 AB 测试和开发者/广告商分析则运行了多个活动集群。外部监控系统会识别故障数量异常的节点，这些节点会从集群中移除，并且修复后的节点会自动地重新加入集群。这些机制虽然都会在不同程度上减少了不可用的持续时间，但仍不能完全隐藏故障。

标准检查点或部分恢复技术的计算成本很高，并且，在（当前）系统中设计实现流式实时地地返回结果给客户端非常难。基于副本的容错机制也将消耗大量资源。考虑到成本，此类技术的预期价值尚不清楚，特别是考虑到节点平均故障时间，以及约 1000 个节点的集群规模和监控数据显示大多数查询（包括批处理 ETL）仍在几小时内完成时。其他研究人员也得出了类似的结论。

但是，我们正在积极致力于改进长时间运行的查询的容错能力。正在评估针对计划的子树添加可选的检查点以及限制重启，但此机制不能以管道方式运行。

V. 查询处理优化

在本节中，我们将描述一些对大多数用例有益的重要查询处理优化。

A. 使用 JVM

Presto 使用 Java 实现并在 Hotspot Java 虚拟机 (JVM) 上运行。从实现中提取尽量好的性能需要利用底层平台的优势和限定性。数据压缩或校验算法等性能敏感的代码可以从特定优化或 CPU 指令中收益。虽然没有应用程序级的机制来控制 JVM 即时 (JIT) 编译器如何生成机器码，但可以构造能够利用 JIT 编译器优化的代码，这类优化代码包括：方法内联、循环展开和内置函数。针对无法生成优化机器码的场景（比如，128位数据运算），我们正在探索使用 Graal（可静态编译执行的 JVM）。

垃圾回收（GC）算法的选择会对应用程序性能产生巨大影响，甚至会影响应用程序实现策略的选择。Presto 使用 G1 收集器，但其对大于特定大小的对象处理不佳。为了限制这些对象的数量，Presto 避免分配大于“humongous”阈值的对象或者缓冲区，并在必要时使用分段数组。由于 G1 中需要维护记忆集结构（remembered set），由此大型和高度链接的对象图也可能引发垃圾回收慢的问题。查询执行的关键路径中的数据结构是基于平面内存数组实现的，依此能够减少引用和对象数量，简化 GC 的工作。例如，HISTOGRAM 聚合将所有分组对应的桶键和数量采用一组平面数组和哈希表存储，而不是为每个直方图维护独立的对象。

B. 代码生成

Presto 引擎的主要性能特征之一是代码生成，目标是 JVM 字节码。这有两种形式：

1. 表达式计算

查询引擎的性能部分取决于计算复杂表达式的速度。Presto 包含一个表达式解释器，可以计算用于测试的任意复杂表达式，但对于计算数十亿行的生产环境来说执行速度较慢。为了加快速度，Presto 生成本地字节码以处理常量、函数调用、变量引用以及延时或短路操作。

2. 针对 JIT 的启发式优化

Presto 为几个关键的“算子”和“算子组合”生成了字节码。生成器利用引擎对计算语义方面的卓越知识来生成字节码，并且此字节码比常规处理循环更适合 JIT 优化。生成器针对三种主要行为：

• 由于引擎在每个单位调度时间均会切换为来自不同任务管道的不同分片，导致已收集的紧密处理循环的监控信息会被其他任务或者查询污染，由此 JIT 将无法基于通用循环的实现执行优化操作。
• 即使在单个任务管道的循环处理中，引擎也会感知每个计算中涉及的类型，并且可以在列上生成展开的循环。消除目标类型差异将导致监控器推断出调度点是单模态的，从而允许其内联虚拟方法。
• 由于每个任务生成的字节码会被编程一个单独的 Java 类，因此 JIT 优化器可以独立地分析每个任务。实际上，JIT 优化器会根据真实的处理数据进一步优化查询生成的自定义程序。这种分析在每个任务中会独立执行，在每个任务处理不同的数据分区的场景中，此机制改进了性能。此外，在任务的整个生命周期中，数据的变化（时间序列数据或日志）会导致生成代码被更新，随之性能也可能发生变化。

生成的字节码也受益于内联的二阶效应。JVM 能够扩大优化范围，扩大计算的自动向量化部分，并可以利用基于频率的基本块布局来最小化分支。CPU 分支预测也变得更加有效。字节码生成改进了引擎将中间结果存储在寄存器或缓存中而不是内存中的能力。

C. 文件格式特性

扫描（scan）算子使用叶子分片信息调用连接器 API 并以页的形式接收列式数据。一个页由块列表组成。每个块对应一列，并采用平面内存方式存储。采用平面内存数据结构对性能是重要的，尤其是对复杂类型。指针追踪、拆箱和虚方法调用会显着增加紧密循环的开销。

Hive 和 Raptor 等连接器会尽可能的利用文件格式的固有特征。某些文件格式能够通过使用文件头或文件尾中的统计信息（例如，最小最大范围等前缀信息和布隆过滤器）快速跳过数据块，Presto 针对这些文件格式提供了自定义读取器。读取器可以将特定格式的压缩数据直接转换为块，引擎能够对块执行高效操作。

图 5 展示了一个页的布局，并且每列都采用了压缩编码方案。字典编码块可以非常有效地压缩具有低基数特性的数据，运行长度编码（RLE）块能够压缩重复数据。几个页也许共享同一个字典，这大大改进了内存效率。ORC文件中的单列可以对整个“条带”（最多数百万行）使用同一个字典。

D. 数据延时加载

Presto 支持数据的延迟物化。此功能可以利用 ORC、Parquet 和 RCFile 等文件格式的列压缩的固有属性。连接器可以生成具有延时加载特性的块（lazy blocks），当单元格被真正访问时，才会读取、解压缩和解码数据。鉴于大部分 CPU 时间花在解压和解码，且过滤器通常具有高选择性，因此这种优化在不经常访问列的场景会非常有效。利用来自生产环境的批处理 ETL 用例样本进行的测试表明，采用了延时加载机制，拉取的数据减少了 78%，单元加载减少了 22%，总 CPU 时间减少了 14%。

E. 基于压缩数据进行操作

Presto 尽可能对来自连接器的压缩数据（字典块及运行长度编码块）直接进行操作。图 5 展示这些块在页中的结构。当执行转换或者过滤功能的页处理器遇到字典块时，其会处理字典中的所有值（或者运行长度编码块中的单个值）。在快速的无条件循环中，引擎会处理整个字典。在某些情况下，字典中存在的值数比块中的行数还要多。在这种情况下，页处理器推测未引用的值将在后续块中使用。页处理器持续跟踪生成的实际行数和字典的大小，这有助于衡量处理字典相对于处理所有索引的是否更有效。如果行数大于字典大小，则处理字典可能更有效。当页处理器在一系列块中遇到了一个新字典，其会启发式地决定是否继续推测。

Presto 在构建哈希表（例如，join 或 agg）时也利用了字典块结构。在处理索引时，算子会采用数组记录每个字典条目对应哈希表位置。如果后续索引中这个条目重复出现，那么只是简单的重复利用这个位置信息而不是去重新计算它。当连续的块共享相同字典时，页处理器会保留这个数组以进一步减少不必要的计算。

Presto 在执行期间也会生成中间压缩结果。例如，join 处理器会尽量生成效率更高的字典块或者运行长度编码块。对于哈希 join，当探测端在哈希表中查询键时，会将值索引记录到一个数组，而不是复制实际数据。算子简单的生成一个字典块，其中索引列表是该数组，字典是对哈希表中块的引用。

VI. 性能

在本节中，展示了性能结果，证明了本文描述的一些主要设计决策的影响。

A. 适应性

在 Facebook 内部，生产环境运行多个不同的连接器，以允许用户处理存储于各种内部系统中的数据。表 1 概述了用于支持第 II 节中概述的用例的连接器及其部署。

为了证明 Presto 如何适应连接器特性，在 30TB 数据规模的基础上，我们比较了来自 TPC-DS 基准测试的查询的运行时间。Presto 能够运行所有的 TPC-DS 查询，但对于此实验，我们选择了一个不需要溢出的低内存特性的查询子集。

我们使用带有 Hive/HDFS 和 Raptor 内部变体连接器的 Presto 0.211 版本。Raptor 是一个为 Presto 设计的无共享存储引擎，采用 MySQL 存储元数据，在本地闪存硬盘上以 ORC 格式存储数据。Raptor 支持复杂的数据组织（排序、分桶和时间列），但对于此实验，数据是随机分区的。 Hive 连接器使用了类似于 Hive Metastore 的内部服务，并访问了与 HDFS 功能相似的远程分布式文件系统中以类 ORC 格式编码的文件。这些连接器变体的性能特征跟公共云提供商上的部署类似。

每个查询都在100个节点的测试集群上以三种配置运行：（1）数据存储于 Raptor，表分片随机分布在各个节点。 (2) 数据存储于 Hive/HDFS，没有统计信息。（3）数据存储于 Hive/HDFS，表和列均具有统计信息。当存在统计信息时，Presto 优化器可以对 join 顺序和 join 策略做出基于成本的决策（CBO）。每个节点都配置了一个运行频率为 2.40GHz 的 28 核 IntelTM XeonTM E5-2680 v4 CPU、1.6TB 闪存和 256GB DDR4 RAM。

图 6 展示了 Presto 查询的运行时间受连接器特征影响很大。在不改变查询和集群配置的情况下，Presto 能够利用其特征适配连接器，这些特征包括：吞吐、延时和统计信息的可用性。还证明了单个 Presto 集群可以同时用作传统的企业级数据仓库（数据必须被加载）和 Hadoop 数据仓库两者之上的查询引擎。为实现更快速的分析结果和低延时的仪表盘，Facebook 的数据工程师经常使用 Presto 对 Hadoop 数据仓库执行探索性分析，然后将聚合结果或者经常访问的数据加载到 Raptor 中。

B. 灵活性

Presto 的灵活性在很大程度上归功于其低延迟的数据 shuffle 机制以及支持高性能处理大量数据的连接器 API。图 7 展示了生产环境所选用例的查询运行时间的分布。只包含了成功的查询，并且这些查询是从存储中读取的真实数据。结果证明，Presto 不仅可以有效服务于具有严格延时要求（20~100ms）的 Web 用例，还可以服务于编程调度的可以运行数小时的 ETL 作业。

C. 资源管理

在多租户集群中，Presto 集成的细粒度资源管理系统，使其在查询之间能够快速移动 CPU 和内存资源，最大限度的提升了资源效率。图 8 展示了来自一个交互式分析集群持续跟踪 4 个小时的 CPU 和并发指标数据。即使需求从 44 个查询的峰值下降到 8 个查询的低点，Presto 的各个 Worker 节点仍能继续保持上平均 90% 的 CPU 利用率。另一个值得注意的是，在新的计算成本低的工作负载到达时，调度器会对其进行优先调度，以保证响应能力。在新查询被接纳后的几毫秒内，通过将整个集群范围的大部分 CPU 分配给新查询来实现这一点。

VII. 工程经验

2013 年以来，在 Facebook 中， Presto 作为一个服务，一直由一个小团队开发和运维。我们观察到，在快速发展的环境中，一些工程哲学通过反馈循环地对 Presto 的设计产生了巨大影响：

可配置的适应性

作为能够执行任意用户自定义计算需求的复杂多租户查询引擎，Presto 不仅必须适应查询特征的差异化，而且还需要组合这些特征。例如，在 Presto 有端到端自适应背压之前，具有慢客户端的少量作业却使用了大量的内存和 CPU，这对同时并发运行的延时敏感作业造成了负面影响。没有适应性，那就需要根据工作负载进行严格分区，并针对每个分区的工作负载分别调整配置参数。生产环境中的查询样式的多样性导致这种方法不会被大面积部署。

易用的仪表盘

Presto 暴露了查询和节点的细粒度性能统计信息。维护了自研的代码库以进行高效的统计收集，并且采用宽内存（flat-memory）以快速计算近似的内存占用量。鼓励设计可观测系统并允许工程师检测和了解其代码的性能也非常重要。自研的代码库使添加统计信息就像注释方法一样简单。因此，Presto Worker 节点输出了约 10,000 个实时性能计数器，收集并存储了每个查询的算子级别的统计信息，这些算子级别的统计信息又可以进一步合并为任务和阶段级别的统计信息。对监控工具的投入，使我们能够在优化系统时以数据为驱动。

静态配置

像 Presto 这样的复杂系统中的算子问题很难找到根本原因并快速缓解。配置属性会以难以推理的方式影响系统性能，我们优先考虑能够理解集群的状态，而不是快速更改配置的能力。与 Facebook 的其他几个系统不同，Presto 尽可能使用静态而不是动态配置。我们开发了自己的配置库，如果有任何警告，则会在启动时崩溃，从而“大声”失败，这包括配置了未使用、重复或冲突的属性。该模型给自身带来了一系列挑战。然而，对于大量的集群和配置集，将复杂性从运维调查转移到部署过程及工具会更加有效。

垂直整合

与其他工程团队类似，针对性能和效率重要的组件设计了自定义库。例如，自定义的文件格式读取器允许我们采用 Presto 本地的数据结构进而避免了转换开销。然而，在一个长期运行的进程中，并且此系统以高并发模式执行任意计算，我们观察到，此时能够轻松地调试和控制库行为的能力同等重要。

考虑最近的一个生产环境问题的例子。Presto采用了 Java 内置的 gzip 库。当调试一系列进程崩溃时，我们发现 glibc 和 gzip 之间的调用本地代码部分会引发内存碎片。对于特定的工作负载组合，这会引发大量的本地内存泄漏。为解决此问题，我们改变了使用库的方式以能够正确的刷新缓存，但在另一个场景，我们甚至为压缩格式编写了自定义库。

自定义库还可以改善开发者效率，通过只实现必要的功能、统一配置管理和支持详细的工具来匹配用例，以减少 bug 的影响面。

VIII. 相关工作

过去十年，在大型数据集上运行 SQL 的系统已经变得流行。每个系统都呈现了一组独特的权衡。对此领域的全面性的调查超出了本论文范围。相反，本论文仅关注该领域一些更需要注意的工作。

Apache Hive 最初是由 Facebook 开发的一种为存储于 HDFS 中的数据提供类 SQL 的接口，通过编译为 MapReduce 或 Tez 作业来执行查询。Spark SQL 是一种基于流行的 Spark 引擎的更现代化的系统，其解决了 MapReduce 的很多限制。Spark SQL 可以在多个分布式数据存储上运行大型查询，并且可以在内存中操作中间结果。但这些系统不支持端到端的管道，并且在阶段间 shuffle 时通常需要将数据持久化到文件系统。尽管这提高了容错能力，但额外的延时导致此类系统不适合交互式或低延时用例。

Vertica、Teradata、Redshift 和 Oracle Exadata 等产品可以不同程度地读取外部数据。然而，它们是围绕内部数据存储构建的，只有当数据已被加载到此类系统中时性能才会最好。一些系统采用 RDBMS 样式和 MapReduce 执行的混合方法，例如，微软的 SQL Server 的 Polybase（针对非结构化数据）和 Hadapt （针对性能）。Apache Impala 可以提供交互式延时，但只能在 Hadoop 生态中运行。相比而言，Presto 与数据源无关。管理员可以采用类似 Raptor 的垂直集成的数据存储部署 Presto，但也可以通过配置 Presto 以低开销模式从各种系统（包括关系/NoSQL 数据库、专有内部服务和流处理系统）中查询数据，即使在单个 Presto 集群。

Presto 建立在“系统”和“数据库”社区开发的创新技术的丰富历史之上。其使用了类似于 Neumann 和 Diaconu 等人描述的技术。参考了关于编译查询计划以显著加快查询处理速度。其尽可能使用 Abadi 等人的技术操作压缩数据及生成压缩的中间结果。其可以从 l`a Vertica 和 C-Store 的多个方案中选择最佳布局，并使用类似于 Zhou 等人的策略通过推理计划属性以最大限度的减少 shuffle 数。

IX. 总结

本论文介绍了一种由 Facebook 开发且开源的 MPP SQL 查询引擎 Presto，其可以快速处理大型数据集。Presto 设计的非常灵活（flexible），通过配置其可以为在各种用例中进行高性能 SQL 处理。丰富的插件接口和连接器 API 使其具有可扩展性（extensible），能够集成多个数据源，并在许多环境中都是有效的。此引擎还设计为具有自适应性（adaptive），可以利用连接器的功能去加速执行，并且可以自动调节读取和写入的并发度、网络 I/O、算子启发式执行，能根据当前查询执行时的系统特性优化调度策略。Presto 的架构使其不仅能够为处理需要极低延时的工作负载提供服务，而且也能够为开销较高且运行时间较长的查询提供有效服务。

Presto 允许像 Facebook 这样的组织部署单个 SQL 系统来处理多个常见的分析用例，并能够轻松查询多个存储系统，同时此系统还可以扩展到1000个节点。其架构和设计方案在众多的大数据 SQL 分析空间中占有一席之地。在 Facebook 和行业中的采用率正在快速增长，并且开源社区持续保持了较高的参与度。