这篇论文是写于2018年，并发表在ICDE 2019上，使用的版本还是0.211版本。当时Presto还没分裂出PrestoDB和PrestoSQL，目前他们的最新版本分别是0.223和331。但是不管怎么变，目前来看其内核还是没有太大的变化，所以这篇论文对于我们了解Presto以及其内核还是有非常大的参考价值的。
由于个人水平有限，文中如果有读起来拗口或者翻译出错的地方，欢迎指正并参考原文。
原论文地址：https://research.fb.com/publications/presto-sql-on-everything/

文章目录

  • 它是一个自适应的多租户系统，可以同时运行数百个资源（主要指Memory、IO、CPU等）敏感型的任务，Presto将这些查询任务分发到数百台worker节点上，从而有效的利用集群资源。
  • Presto是可扩展的联邦查询设计架构，允许管理人员在集群中通过一个查询处理多个不同数据源的数据。这样降低了集成多个查询系统的复杂度。
  • 它非常的灵活，可以通过配置不同的配置项提供对不同用例的支持。
  • Presto就是为高性能而生的，集成了多种优化措施，包括代码生成时的优化。多个查询共用一个长期存活的JVM进程，这样可以减少响应时间，但是这些都需要查询调度以及资源管理和隔离等措施的支持。

这篇论文的主要作用就是向大家介绍Presto引擎的设计架构并且讨论特定的优化措施，以及如何平衡这些优化措施来达到上述特性。

Ⅱ. 使用示例

在Facebook，运行着许多的Presto集群（多达上千个节点）用来支持不同的使用场景。这部分会列举4种不同的应用案例。

A. Interactive Analytics（交互式分析）

Facebook内运行着一个庞大的多租户数据仓库，一些业务部门或个别团队会共享其中一小部分托管的集群。其数据存储在一个分布式文件系统之上，而元数据则存储在单独的服务中，这些系统分别具有HDFS和Hive Metastore服务类似的API。我们称之为’Facebook data warehouse’，并且通过类似于Presto ‘Hive’ Connector的组件来进行文件的读写。
Facebook的工程师和数据科学家经常会检索少量的数据(50GB-3TB的压缩数据)，用来验证假设，并构建可视化的数据展板。这些用户通常会使用查询工具、BI工具或Jupyter notebooks来进行查询操作。各个群集需要支持50-100个具有各种查询模型的操作并发执行，并且需要在数秒或数分钟内返回结果。这些用户通常并不关心查询所使用到的硬件资源，但是对查询时间却相当敏感。而对于某些探索性的查询，用户可能并不需要获取所有的查询结果。通常在返回初始结果后，查询就会被立即取消或者用户会通过LIMIT来限制系统返回的结果。

B. Batch ETL （批量ETL）

上面我们介绍到的数据仓库会使用ETL查询任务定期填充新的数据。查询任务通常是通过一个工作流系统依次调度执行的。Presto支持用户从历史遗留的批处理系统迁移ETL任务，目前ETL查询任务在Facebook的Presto工作负载中占了很大一部分。这些查询通常是由数据工程师开发并优化的。相对于Interactive Analytics中涉及的查询，它们通常会占用更多的硬件资源，并且会涉及大量的CPU转换和内存（通常是数TB的分布式内存）密集型的聚合操作以及与其他的大表连接操作。因此相对于资源利用率以及集群吞吐量来说，查询延迟显得没那么重要。

C. A/B Testing （A/B测试）

Facebook使用A/B测试，通过统计假设性的测试来评估产品变更带来的影响。在Facebook大量的A/B测试的基础架构是基于Presto构建的。用户期望测试结果可以在数小时之内呈现（而不是数天），并且结果应该是准确无误的。对于用户来说，能够在交互式延迟的时间内（5~30s），对结果数据进行任意切分来获得更深入的见解同样重要。而通过预处理来聚合这些数据往往很难满足这一需求，因此必须得实时进行计算。生成这样的结果需要关联多个大型数据集，包括用户、设备、测试以及事件属性等数据。由于查询是通过编程方式实现的，所以查询需要被限制在较小的集合内。

D. Developer/Advertiser Analytics（开发者/广告主分析）

为外部开发者和广告客户提供的几种自定义报表工具也都是基于Presto构建的。Facebook Analytics就是其中一个实际案例，它为使用Facebook平台构建应用程序的开发人员提供了高级的分析工具。这些工具通常对外开放一个Web界面，该界面可以生成一组受限的查询模型。查询需要聚合的数据量是非常大的，但是这些查询是有目的性的，因为用户只能访问他们的应用程序或广告的数据。大部分的查询包括连接、聚合以及窗口函数。由于这些工具是交互式的，因此有非常严格的时间限制（约50ms~5s）。鉴于用户的数量，集群需要达到99.999%的可用性，并且支持数百个并发查询。

Ⅲ. 架构概览

一个Presto集群需要由一个Coordinator以及一个或者多个Worker节点组成。Coordinator主要负责接收查询请求、解析语句、生成计划、优化查询以及查询调度。Worker节点主要负责查询处理。如下所示的即为Presto架构：Presto: SQL on Everything（全文翻译） - 图1
图1:Presto架构
客户端向Coordinator发送一个包含sql的http请求。Coordinator接收到这个请求，会通过评估队列策略，解析和分析sql文本，创建和优化分布式执行计划来处理请求。
Coordinator将执行计划分发给Worker节点，接着Worker节点开始启动tasks并且开始枚举splits，而这些splits是对外部存储系统中可寻址的数据块的一种隐晦处理。Splits会被分配给那些负责读取数据的tasks。
Worker节点运行这些tasks来处理从外部存储系统获取的splits，以及来自于其他Worker节点处理过得中间数据。Worker节点之间通过多任务合作机制来并发处理来自不同查询的tasks。任务尽可能的以流水线的方式来执行，这使得数据可以在tasks之间进行流动。对于某些特定的查询，Presto能够在处理完所有数据之前就返回结果。中间数据以及状态会尽可能地存储在内存中。当在节点之间对数据进行shuffle时，Presto会调整缓冲区来达到最小化的延迟。
Presto被设计成可扩展的，并提供通用的插件接口。插件可以提供自定义的数据类型、函数、访问控制、事件监听策略、排队策略以及属性配置。更重要的是插件还提供了connector，这使得Presto可以通过Connector API和外部的存储系统进行通信。Connector API主要包含如下四个部分：Metadata API、Data Location API、Data Source API以及Data Sink API。这些API可以帮助在分布式查询引擎中实现高性能的connectors，开发者已经为Presto社区提供了数十个connector，而且我们也注意到了一些专有的connectors。

Ⅳ. 系统设计

这一部分将介绍Presto引擎的一些关键设计和特性。首先，我们会介绍Presto支持的SQL方言，以及查询从客户端到分布式执行的整个生命周期。接着会介绍Presto中支持的多租户的资源管理机制。最终，会简短的讨论一下容错机制。

A. SQL 方言

Presto严格遵守标准的ANSI SQL规范。虽然Presto并未实现所有的SQL语义，但是所提供的功能都会尽可能的符合规范。为了提高可用性，我们对SQL语义做了一些精心的扩展。例如，在ANSI SQL中实现map、array等操作是相当复杂的。为了简化对这些数据类型的处理，Presto在语法上支持匿名函数（lamada表达式）并且内置了一些高阶的函数（如transform,filter,reduce）。
###B. 客户端接口，SQL解析以及查询计划1）Client接口：Coordinator对客户端提供了REST API并且配备了命令行接口。Presto同样配备了JDBC客户端，以此来兼容大量的BI工具，包括Tableau和Microstrategy。
2）语法解析：Presto使用基于ANTLR的解析器将sql转换成相应的语法树。解析器通过这棵语法树来确定类型、解析函数和作用域，并且抽取如子查询、聚合以及窗口函数等逻辑部分。
3）逻辑计划：逻辑计划器使用语法树和分析信息来生成计划节点的树形编码的中间表示形式。每一个计划节点都表示一个物理或逻辑操作，而计划节点的子节点则为当前节点提供输入数据。逻辑计划器生成的节点是纯粹的逻辑节点，并不包含任何执行信息。例如下面的一个简单查询：
SELECT
orders.orderkey, SUM(tax)
FROM orders
LEFT JOIN lineitem
ON orders.orderkey = lineitem.orderkey
WHERE discount = 0
GROUP BY orders.orderkey
1234567
如上查询生成的逻辑计划如下所示：Presto: SQL on Everything（全文翻译） - 图2
图2:逻辑计划

C. 查询优化

计划优化器会将逻辑计划转换成能够表示有效执行策略的物理结构。转换过程是通过贪婪地评估一组转换规则，直到达到某一特定目标。每个规则都会匹配一种模式，该模式可以与查询计划的子树匹配，并确定是否需要转换。转换的结果是一个逻辑上等价的子计划，该子计划会替换树中满足转换规则的节点。Presto包含了一些通用的的优化规则，如谓词和限制下推、列裁剪以及去相关（decorrelation）。
我们正在使用基于成本的计划评估，来增强优化器以对搜索空间进行更全面的探索，该技术是从Cascades框架引入的。而目前Presto已经支持了两种基于基于成本的优化，这些优化策略考虑了表和列的统计信息，如关联策略选择以及关联重排。这里我们只会介绍优化器的部分功能，而详细的优化策略已经超出了本文的范围。
1）数据分布：Connector提供了数据分布的API，而优化器可以很好的利用数据的物理布局。Connector可以提供数据位置信息以及分区、排序、分组和索引等信息。Connector可以返回一张表的多个物理数据分布情况，每部分的数据具有不同的属性， 优化器可以从中选择对当前查询最有效的数据。该项功能已经被很好的运用在 Developer/Advertiser Analytics的案列中；该系统的超级管理员们可以通过获取数据的物理分布来优化不同的查询。在随后的部分中，我们将看到Presto引擎是如何利用这些属性的。
2）谓词下推:优化器可以与Connector一起使用，用来确定下推的范围以及等效的谓词下推，从而提高过滤效率。
例如，在Developer / Advertiser Analytics的案列中，在MySQL分片之上构建了专有的Connector。Connector将存储在Mysql实例上的数据划分成小的数据分片，并且将下推范围以及关键谓词应用于这些数据分片，以此来保证从Mysql中读取满足需求的数据。如果存在多种数据分布，引擎会选择已在谓词列上构建索引的那部分数据。在Developer / Advertiser Analytics的案列中，基于索引的高效过滤是非常重要的。而在Interactive Analytics和Batch ETL的案例中，Presto会利用Hive connector中的分区修剪和文件格式的特性以类似的方式提查询高性能。
3）节点间并行: 确定执行计划中的哪些部分可以在Worker节点之间并行处理，也属于优化工作的一部分。执行计划中可以被并行处理的部分称之为stage，每个stage都会被分配给一个或多个task，而每个task对不同的输入数据集执行相同的计算。引擎在stages之间进行shuffles以实现数据交换。Shuffles会增加了延迟、耗尽缓冲区内存并造成CPU的高负载。因此，优化器需要慎重考虑计划中被引入的shuffles总数。如图3所示，将计划划分成不同的stage，并通过shuffles连接不同的stage。
数据分布属性:优化器可以使用数据的物理布局来最小化shuffle的次数。这在A/B Testing 的案列中非常有用。在该场景下，几乎每一个查询都需要通过连接大量的数据才能产生结果。考虑到参与连接的两张表都是作用于同一字段，引擎会利用该特点，将同时位于本地的列数据进行关联从而消除耗资源的shuffle操作。
如果Connector向外提供了连接列的数据分布情况，那么参与连接的列会被标记成索引。这样，优化器就能确定本地的基于索引嵌套循环的连接是否是一个好的优化策略。结合生产数据的存储方式（键值存储或其它方式的存储），这种优化措施可以高效的应用于数据仓库中的标准化数据。在Interactive Analytics的案列中该优化措施会被经常用到。
节点属性：如Connector一样，计划树中的节点也包含了其输出数据的各种属性信息（如分区，排序，分桶以及分组等数据特性）。这些节点还包含了一些必要和首选的属性，在进行数据的shuffle时这些信息都可以被考虑在内。多余的shuffle会被删除，但是在其他的一些情况下，可以通过更改shuffle的属性来减少必要的shuffle数量。Presto会贪婪地选择分区，而这些分区将满足尽可能多的必要属性以减少shuffle数量。这意味着优化器可以选择在较少的列上进行分区，而这在某些情况下会导致更多的分区倾斜。例如，应用于图3中计划的优化措施，将导致多个stage合并到单个stage中去执行。Presto: SQL on Everything（全文翻译） - 图3
图3：图2中示例的分布式计划
4）节点内并行：优化器会使用类似的机制来确定执行计划的各个阶段中哪些部分可以通过单节点上的多线程来提高执行效率。与Worker节点间的并行处理相比，节点内的并行处理效率会更高，因为节点内的延迟会更低，并且状态（如哈希表和字典等）也可以在线程之间进行高效的共享。添加节点内的并行可以显著的提高执行效率，特别是对于那些并发限制了下游阶段吞吐量的查询：

  • 在Interactive Analytics的案例中，会涉及到许多简短的一次性查询，用户通常不会花时间来优化这些查询。这样就会导致分区倾斜时常发生，部分是由于数据的某些内在属性以及常见的查询模式导致的（如将用户按国家进行分组，同时也过滤掉一小部分国家）。这通常表现为大量数据的哈希分区只会作用于少量的节点上。
  • 在Batch ETL的案例中，作业任务经常不进行数据过滤或者很少的过滤，就直接对大量的数据集进行转换处理。在这些情况下，计划树中较高层级的节点所涉及到的Worker节点数可能较少，以至于无法快速处理由叶子阶段（leaf stage）所产生的数据。这会涉及到任务调度，后面会介绍。

在如上的两种情况下，在每个Worker节点上通过多线程处理可以在某种程度上缓解并发的瓶颈。引擎可以通过多线程来处理一个序列的算子（也称为管道）。如图4所示，优化器通过并行处理来优化连接操作。Presto: SQL on Everything（全文翻译） - 图4
图4：相对于图3中优化后的示例

D. 调度

Coordinator通过分发可执行的task的方式，向Woker节点分配执行计划的各个stage，这些可执行的task可以被看作单个处理单元。接着，Coordinator将一个stage的task与其他stage的task通过shuffles相连接，从而形成一个树形的处理链路。只要各个stage都是可用的，数据就会在stage之间流转。
一个task的内部可能会存在多个管道（pipeline）。而一个pipeline又是一连串的算子（operator）组成的，而每一个operator都会对数据进行单独的处理。例如，执行哈希联接的任务必须包含至少两个pipeline；一个用于构建哈希表（build pipeline），而另一个则用于在探测端对流入的数据进行连接操作（probe pipeline）。当优化器认为pipeline的部分操作可以通过局部的并行处理来优化时，优化器会对pipeline进行拆分并单独对这部分操作进行并行化处理。图4中展示了build pipeline被划分成两个pipeline，一个用于扫描数据，而另一个则用于构建局部哈希表。两个pipeline之间则通过local shuffle连接起来。
为了执行一个查询，Presto引擎会制定一个两阶段的调度策略。首先，需要确定各个stage之间的调度顺序。其次，需要确定有多少task会被调度，以及这些task需要被分配到哪些Woker节点上。
1）Stage 调度：在这一调度阶段，Presto支持两种调度策略：一次性调度（all-at-once）以及阶段调度（phased）。一次性调度所有的stage并发执行可以最小化CPU时钟开销；一旦数据可用，就会被立即进行处理。这种调度策略有利于那些对延迟敏的场景，如Interactive Analytics，Developer/Advertiser Analytics以及A/B Testing的案例。分阶段调度的策略会首先确定有向数据流图中所有必须紧密连接的单元，这些单元必须同时启动以避免死锁，并且需要按照拓扑顺序来执行。例如，如果在阶段调度模式下执行hash-join，则在构建哈希表之前，都不会调度执行hash-join左侧的任务。这大大的提高了Batch Analytics案列中的内存使用效率。
当调度器根据调度策略确定某个stage应该被调度时，它会开始将该stage内的task分配给Worker节点来执行。
2）Task调度：Task调度器通过检查计划树会将stage划分为叶子阶段（leaf stages）或中间阶段（intermediate stages）。叶子阶段会通过connector来读取数据；而中间阶段仅处理其他阶段产生的中间结果。
叶子阶段（leaf stage）：在叶子阶段，Task调度器在将task分配给Worker节点时，会考虑网络以及连接器所带来的影响。例如，无共享的部署方式会要求Worker节点和存储节点在一起。在这种情况下，调度器会使用connector的Layout API来确定数据的分布方式从而确定task的执行位置。在A/B Testing的案列中，需要可预测的高吞吐量以及低延迟的数据读取，而Raptor connector可以满足这些需求。Raptor是针对Presto优化的存储引擎，采用无共享架构，可将ORC文件存储在闪存盘上，并在MySQL中存储其元数据。
分析显示，我们整个生产集群上的大部分CPU时间都用于对那些从connector读取的数据进行解压缩，解码，过滤和转换。而这些工作是可以被高度并行执行的，并且在尽可能多的节点上运行这些stage通常会产生最短的挂墙时间（wall time）。因此，如果没有加以限制，并且源数据可以被划分为足够多的split，叶子阶段的task会在集群中所有的worker节点上执行。对于以共享存储模式(即所有数据都是远程存储的)运行的Facebook数据仓库而言，群集中的每个节点通常都会参与叶子阶段的任务处理。但是这种执行策略是网络敏感性的。
调度器还可以使用插件提供的层次结构来推理网络拓扑结构，从而优化读取效率。在Facebook，网络受限的部署方式可以利用这种机制帮助Presto引擎优先读取机架本地的数据，而不是远程机架的数据。
中间阶段（Intermediate Stages）：中间阶段的task可以放在任意worker节点来运行。但是，Presto引擎仍然需要确定每个stage应计划执行多少个task。而这些是基于Connector的配置，执行计划中的属性，被处理数据的分布以及其他部署配置所决定的。而在某些情况下，执行引擎可以在执行任务期间动态更改任务数。
3）Split调度：当叶子阶段的task开始在worker节点上执行时，该节点会接收一个或者多个data split。一个split包含的信息因Connector而异。当从分布式文件系统读取数据时，split可能包含文件路径和当前文件的偏移量。对于Redis这种键值存储系统来说，一个split会包含表信息，键和值格式以及要查询的主机列表等。
处于叶子阶段的每一个task需要分配到一个或者多个data split之后才能被执行，而处于中间阶段的task始终是可运行的，并且仅在task被终止时或完成其所有上游task后才停下来。
分配分片（Split Assignment）：一旦tasks被分配到worker节点之后，Coordinator开始将data split分配给这些tasks。Presto会请求Connector枚举小批量的data split，然后将它们延迟（lazy）分配给这些tasks。这是Presto的一个重要功能，它带来了许多的好处：

  • 将查询响应时间和Connector枚举大量的data split的时间解耦。例如，Hive-connector可能需要数分钟来枚举分区并列出每个分区目录中的文件。
  • 可以在不处理完所有数据的情况下就开始生成查询结果（例如，简单地使用过滤器来查询数据），这些查询操作通常会被快速的取消或者在满足LIMIT子句时提早完成。在 Interactive Analytics的案列中，通常是在所有data split被枚举完之前完成查询。
  • Worker节点上都维护着一个将要被处理的data split队列。Coordinator会简单的将data split分配给那些拥有最短split队列的task。 保持这些队列尽量的小，可以使系统能够适应处理不同split带来的CPU成本差异以及worker节点间的性能差异。
  • 允许在执行查询时，不必将所有的元数据都保存在内存中。这对于Hive-connector是至关重要的，因为在执行查询时Hive-connector可能会访问数百万个data split，这会导致Coordinator上所有可用的内存很快的被消耗殆尽。

对于运行在Facebook内Hive兼容的数据仓库上的Interactive Analytics 和 Batch ETL的案例，这些特性会显得格外的有用。 但是值得注意的是，data split的懒加载可能会导致精确估算和报告查询进度变得困难。

E. 查选执行

1）本地数据流：一旦data split分配给线程后，它会在driver中循环执行。Presto的driver中的循环比流行的Volcano递归迭代器模型更复杂，但却提供了重要的功能。它更适用于多任务的协作处理，因为算子（operator）可以在线程生成之前就迅速进入已知的状态，而不是无限地阻塞下去。此外，driver可以在不需要额外的输入文件的情况下，通过移动operator之间的数据使每个数据量子的执行效率达到最大化（如恢复资源密集型的计算或爆炸性转换的计算）。循环的每次迭代都会在operators之间移动数据驱使查询不断的执行。
drive循环中操作的数据单元称之为page，而一个page为某一列一串行值编码后的数据。Connector的Data Source API在传入一个data split之后会返回若干个page。如图5所示得为内存中page的结构。drive中的循环会在operators之间移动page数据，直到调度完成或operator无法继续执行为止。
2）Shuffles：Presto的设计原则旨在最大化资源利用率，同时最小化端到端延迟，而Presto节点间的数据流机制正反映了这种设计原则。Presto基于HTTP协议，使用内存缓冲区在节点间交换中间结果数据来完成shuffles。tasks产生的数据存储在内存缓冲区中，以供其他worker节点来消费。Worker间使用HTTP长轮询从其他worker节点请求中间结果数据。Worker节点上的服务会一直保留中间数据，直到当前worker节点接收到其它worker节点在请求下一段数据时带上一次响应中的token为止。这样的确认机制是隐含在传输协议中完成的。长轮询机制最大程度地缩短了响应时间，尤其是在传输少量数据时更明显。相对于其他系统将shuffle数据持久化到磁盘，Presto提供的shuffle机制带来了更低的延迟，这也使Presto对诸如Developer/Advertiser Analytics这样延迟敏感型的案列提供了很好的支持。
引擎会调整并行度以维持输出和输入缓存的目标利用率。满负荷的使用输出缓存，会导致对split的计算停止并耗尽宝贵的内存资源，而未充分利用的输入缓存也会增加不必要的处理开销。
引擎会持续监控输出缓存的利用率。当利用率居高不下时，它会减少可被执行的split数量，从而有效的降低并发度。该措施可以提高网络资源共享的公平性。当客户端（用户端或其它woker）的处理速度跟不上数据的产生速度时，这也是一种重要的优化措施。如果没有这样的措施，那些处理速度较慢的客户端在运行复杂的多阶段查询时会一直持有数十GB的缓存。在 Interactive Analytics的案列中，Bi或查询工具通过慢速的连接下载少量的数据（通常10~50MB）时，这样的情况会经常出现。
在接收端，引擎会监控每个请求传输的平均数据量来计算目标HTTP请求的并发度，这样可以保证输入缓存得到充分的利用并且不超出输入缓存的的容量。这也可以让上游任务在输入缓存填满时减慢执行速度。
3）写：ETL作业通常会产生必须写入其他表的数据。在远程存储环境中写性能的重要驱动因素就是写操作的并发度（如通过Connector Data Sink API来控制数据写线程的数量）。
在使用Amazon S3作为存储的Hive-connector的案例中，对S3的每次并发写入都会创建一个新文件，而数百次的小批量数据的写则会产生许多小文件。除非后续会合并这些小文件，否则在读取文件时会导致难以接受的高开销（许多慢速的元数据操作以及延迟会限制读取性能）。但是，使用过低的并发会使聚合写的吞吐量降低到无法接受的水平。Presto再次采用一种自适应的策略，当引擎认定stage中写操作生成的数据超出了缓存利用率的阈值（可配置的阈值）时，会通过在更多的Worker节点上添加任务来动态增加写程序的并发度。在写操作很重的Batch ETL案列中，这是一种重要的优化方式。

F. 资源管理

Presto非常适合多租户使用，而关键的因素就在于它内置了一个细粒度资源管理系统。这使得一个集群可以同时执行数百个查询，并最大程度地利用CPU，IO和内存资源。
1）CPU调度：Presto是针对整个集群的吞吐率进行优化的，如整合整个集群的CPU来处理数据。此外，本地（节点层面）的调度器针对成本较低且低周转时间的查询进行了优化，以及在具有类似CPU需求的查询之间公平的共享CPU资源。一个task的CPU使用情况是通过累加分配给每个split的处理线程的CPU时间得来的。为了最大程度地减少协调开销，Presto会在任务级别跟踪CPU资源使用情况，并在本地做出调度决策。
Presto在每个Worker节点上调度许多并发任务，以实现多租户，并且使用了一个协作的多任务处理模型。任意给定的split只能在一个线程上最多运行1秒钟，之后必须放弃该线程并返回队列中。当输出缓存已满（下游阶段无法足够快地消耗数据），输入缓存为空（上游阶段可能无法足够快地生成数据），或系统内存不足时，本地调度器会在当前任务CPU运行时间（指运行1秒中）到达之前切换到另一个任务。这样释放了那些正在处理split的线程，帮助Presto最大限度的利用CPU，从而适应不同的查询。我们之前介绍到的案列都受益于这种细粒度的资源管控。
当一个split放弃当前的线程时，查询引擎会确定一下个要被执行的task（与一个或多个split相关联）。Presto并没有提前预测完成一个新的查询所需的资源，而只是使用任务的总CPU时间，将一个多级反馈队列划分成五个层级。随着task累积的CPU时间越来越多，它将被划分到更高的级别。每个级别都指定了一个可配置的CPU时间。在实际应用中，要在任意负载下公平的协调多任务的处理是相当有挑战性的。不同的split使用的IO和CPU的差异性相当的大，即便在相同的任务中也是如此，而复杂的函数（如正则表达式）相对于其他的split可能会消耗更多的线程时间。而某些Connector不提供异步的API，因此有的工作线程会被保留数分钟。
在处理这些约束条件时，调度程序必须是自适应的。系统提供了低成本的信号量，因此在一个operator中可以暂停长时间的运算。如果operator中的停止时间超过特定的值，调度器会将实际的线程时间计入当前task，并且会暂时调低这个任务将来的执行频率。这种自适应的特性允许我们在Interactive Analytics 以及Batch ETL的案例中处理不同类型的查询，其中presto可以为消耗最低资源的查询提供更高的优先级。我们也可以这样来理解，用户只希望低成本的查询可以被快速执行，而不必担心高成本的查询的运转时间。同时运行更多的查询，即使以牺牲更多上下文切换为代价，也可以减少总的排队时间，因为低成本的查询会被快速的执行并退出。
2）内存管理：在Presto这样的多租户系统中，内存构成了资源管理的主要挑战之一。这一节我们会介绍presto在集群中的内存分配机制。
内存池：Presto中的所有的内存资源都要被划分成相应内存池中的用户内存（user memory）、系统内存（systerm memory）以及保留内存（reserve memory）。User memory可以理解为系统或输入数据所使用的内存（内存的使用情况和输入数据的基数成正比）。而systerm memory则是伴随着查询执行而产生的（如shuffle缓存等），和查询类型以及输入数据量没有太大的关系。
Presto对user memory以及total memory（user+system）是做单独限制的，如果查询超过了全局的内存限制（所有worker节点使用的内存的总和）或者超过了单个worker的限制，查询都会被kill掉。当Worker内存不足时，presto会通过停止tasks来阻塞内存分配。Total memory的限制通常设置得比user memory限制大得多，在生产环境中只有少部分的查询会超过total memory的限制。
对于一个查询而言，单worker设置的user memory限制和全局的user memory限制通常是不同的；这样可以最大限度的允许内存使用的倾斜。比如，有一个500个节点的集群，每个节点有100GB的user memory，这样每个查询在集群中最多可以使用多达5T的内存。在这种情况下，将总的内存分配给10个并发执行的查询。但是，如果我们允许2：1的倾斜（即一个查询子部分的内存消耗是内存消耗中位数的两倍），那么每个Worker节点上的查询内存限制需要被设置成20GB。这意味着在不耗尽可用节点内存的情况下，仅可以保证5个查询能够运行。（译者注：一个Worker100G，分给10个查询那么每个查询分到10G，如果允许2：1的倾斜，那么单个查询的内存需要限制20G，那么对于100G的Worker来说最多只能运行5个查询）
但值得注意的是，在InteractiveAnalytics和BatchETL的案列中，我们能够在500个节点的集群上运行远超过5个的查询。鉴于这些集群中的查询在其内存使用特征（倾斜，分配速率和分配的时间点）方面存在巨大差异，因此这5个查询在任何时间点都不可能在一个Worker上使用到最大限制的内存。因此，当Worker节点内存不足时，只要有相应的机制能够保证集群的健康，那么就可以放心的过量的使用内存。Presto提供了两种保障机制：溢出（spilling）和预留池（reserved pool）。
译者注：0.217中system memory pool被移除了
Spilling：当Worker节点内存不足时，presto会对满足条件的task按照执行时间进行升序排序，并对这些task进行内存回收。当有足够的内存满足最后一次请求调用时，则停止内存回收。内存回收是通过将内存溢出到磁盘来完成的。Presto支持对哈希连接以及聚合操作的溢出。但是在Facebook的部署环境中，我们未开启溢出功能。因为集群的分布式内存足够的大，通常是好几TB，用户通常对预期的查询延迟感到满意，而且溢出到本地磁盘会增加硬件成本的开销（尤其是在Facebook的共享存储的部署模式下）。
Reserved Pool：如果节点内存不足，并且集群没有配置为Spilling，或没有剩余的可撤销内存，reserved pool机制可以保证集群不被阻塞。每个节点上的查询内存池会进一步被细分为两个池：general pool和reserved pool。当Worker节点上的general pool耗尽时,那么在worker节点上的占用内存最多的那查询会在整个集群中被提升到reserved pool中。在这种情况下，该查询所消耗的是reserved pool中的内存，而不在是general pool中的内存。为了避免死锁，在整个集群中同时只有一个查询可以在reserved pool中执行。如果Worker节点上的general pool内存已用完，而reserved pool也已经被占用，那么该Worker节点上其他task的所有内存相关的请求将被阻塞。运行在reserved pool中的查询会一直占用该pool直到其执行完成，这个时候，群集将停止阻塞之前所有未完成的内存请求。这在某种程度上看起来有点浪费，因此必须合理的调整每个Worker节点上reserved pool的大小，以满足在本地内存限制下运行查询。当某个查询阻塞了大部分的Worker节点时，集群也可以配置成kill掉这个查询。

G. 容错

目前，Presto可以通过低级别的重试从临时的错误中恢复。但是，截至2018年末，Presto对Coordinator或Worker节点的崩溃，没有任何内置的有意义的容灾措施。Coordinator的故障将导致集群的不可用，而Worker节点的崩溃将导致该节点上所有正在执行的查询失败。目前，对于这样的错误，Presto需要依靠Client端重新提交失败的查询来解决。
目前，在Facebook的生产环境中是通过额外的机制保证某些特定场景下的高可用。在Interactive Analytics 和Batch ETL 的案例中运行着一个备用的Coordinator，而在A/B Testing以及Developer/Advertiser Analytics 的案列中，运行着多个集群来保证高可用。外部监控系统会识别那些产生异常故障数量的节点并将它们从集群中剔除，而被修复的节点会重新加入集群。以上的措施都是在某种程度上降低服务不可用的时间，但是无法完全屏蔽故障的发生。
常用的check pointing 以及部分恢复机制通常是非常消耗计算资源的，并且很难在即席查询系统中实现。基于复制策略的容错机制通常也是相当耗资源的。考虑到这样的成本开销，这种技术的预期价值还尚不明确，尤其是在考虑到节点平均故障时间时，如在Batch ETL的案列中，集群的节点数达到了数千个并且大部分的查询都是在数小时之内完成的。在其他的研究中也得出过类似的结论。
但是，我们正在积极致力于提高长时间运行的查询的容错能力。我们正在评估添加可选的check pointing以及重试那些无法以流水线方式运行的计划子树。

Ⅴ. 查询过程优化

A. 使用JVM

Presto是通过Java语言实现的，并且运行在Hotspot Java虚拟机上。为了在Presto的实现中获得最佳的执行性能，需要发挥基础平台的优势并且避免其劣势。性能敏感型的代码（如数据压缩或校验和算法）可以受益于特定的优化或CPU指令。尽管没有应用级的机制来控制JVM即时编译器（JIT）如何生成机器码，但是可以对代码进行结构化，以便其可以利用JIT编译器提供的优化方式，如方法内联，循环体展开以及内联函数。我们正在探索当JVM无法生成最佳机器码的情况下（如128位的数学运算）使用Graal。
垃圾收集（GC）算法的选择会对应用程序性能产生巨大的影响，甚至可能影响应用程序的实现方式。Presto使用G1收集器，该收集器在处理大于特定大小的对象时性能较差。为了限制这类对象的数量，Presto避免分配大于“humongous”阈值的对象或缓存，并在必要时使用分段数组。由于G1需要维护对象集合的结构（remembered sets tructures），所以大型和高度关联的对象图可能会存在一些问题。查询执行路径上的关键步骤的数据结构是通过扁平化得内存数组来实现的，目的就是为了减少引用以及对象数量，从而使垃圾回收变得相对轻松一点。例如，在HISTOGRAM聚合中，会在一个扁平的数组或哈希表中存储所有组中的桶键（bucket keys）以及对象计数，而不是为每一个histogram维护独立的对象。

B. 代码生成

引擎的主要性能特征之一，就是生产JVM字节码。这有如下两种表现形式：
1）表达式求值：查询引擎的性能部分取决于它对复杂表达式的求值速度。Presto提供了一个表达式解释器，该表达式解释器在我们的测试场景中可以对任意复杂的表达式进行求值，但对于数十亿行的生产数据的求值还是太慢了。为了加速表达式求值，Presto生成字节码来处理常量、函数调用、对变量的引用以及延迟或存在短路效应的操作。
2）针对JIT优化器：Presto会为一些关键算子（operators）和算子组合生成字节码。字节码生成器利用引擎在计算语义方面的优势来生成更易于JIT优化器进行优化的字节码。生成器主要会针对如下三种行为：

  • 由于引擎会在不同任务管道（task pipeline）的不同split之间进行切换，因此JIT无法优化那些基于常见的循环的实现。这是因为，在查询的过程中从循环中得到的分析信息可能会被其他的task或查询所破坏。
  • 即使在单个task pipeline中的循环处理，引擎也知道每次计算所涉及的类型，并且可以在列上生成展开的循环。消除循环体中的类型差异会让分析器得出如下的结论：单一调用可以对虚拟方法进行内联。
  • 由于为每个任务生成的字节码被编译到一个单独的Java类中，因此JIT优化器可以分别对每一个字节码进行概要分析。实际上, JIT优化器会为处理实际数据的代码做进一步的调整。这种基于分析结果的调整可以在每一个独立的task中展开，从而提高了每个task在处理数据的不同分区时的性能。此外，性能分析可以在任务的整个生命周期中随着数据（如，时间序列数据或日志数据）的变化而变化，从而使生成的代码也随之变化。

生成的字节码还得益于内联所带来的二次效应。JVM能够扩大优化范围，自动向量化大部分的计算，并且可以利用基于频率的基本块布局来最大程度地减少分支。这样使得CPU分支预测也变得更加有效。字节码生成提高了引擎将中间结果存储在CPU寄存器或CPU缓存中而不是内存中的能力。

C. 文件格式的特性

扫描算子（Scan operator）使用叶子阶段split的信息来调用Connector API，并以Pages的形式接收列数据。一个page由一个block列表组成，而每个block是具有扁平内存表示形式的列。使用扁平内存的数据结构对性能非常重要，尤其是对那些复杂的数据类型。在紧凑的循环体内，指针跟踪，拆箱和虚拟方法调用都增加了大量的开销。
Hive和Raptor这类connectors会尽可能的使用特定的文件格式。Presto配有自定义的reader，可以通过使用文件页眉/页脚中的统计信息（如页眉中保存的最小-最大范围和布隆过滤器），来高效地过滤数据。Reader可以直接将某些形式的压缩数据直接读成blocks，从而让引擎可以有效地对其进行处理。
图5显示了一个page中列的布局方式，其中每一列都有其对应的编码方式。字典编码的块（DictionaryBlock）在压缩数据的低基数部分非常高效，游程编码的块（run-length encoded block，RLEBlock）对重复的数据进行压缩。多个page可以共享一个字典，这可以大大提高内存利用率。ORC文件中的一列可以为整个“stripe”（最多数百万行），使用单个字典。Presto: SQL on Everything（全文翻译） - 图5
图5：一个page中不同的block类型

D.数据的懒加载

Presto支持数据的惰性物化（lazy materialization）。此功能可以利用ORC，Parquet和RCFile等文件格式的列压缩特性。Connector可以生成惰性的blocks，仅当实际访问时才读取，解压缩和解码数据。鉴于大部分的CPU时间都花在了解压缩和解码上，并且查询的数据都是有目的性的，因此这种优化在不经常访问列的情况下非常有效。在Batch ETL的案列中，对生产环境中负载样本进行的测试表明，懒加载将提取的数据减少了78％，将单元加载的数据减少了22％，将总CPU时间减少了14％。

E. 操作压缩数据

Presto会尽可能的处理来自Connector的压缩数据（即Dictionary和RLE block）。图5中显示了page内这些block的结构。当page处理器在计算转换或过滤时遇到dictionary block，它将处理 dictionary 内的所有的值（或RLE的块中的单个值）。这允许引擎以快速的无条件循环处理整个dictionary。在某些情况下，dictionary中存在的值多于block中的行。在这种场景下，page处理器推测未引用的值将在后续的block中使用。Page处理器会持续跟踪产生的实际行数和dictionary的大小，这有利于对比处理索引和处理dictionary的效率。如果行数大于dictionary的大小，则处理dictionary的效率可能更高。当page处理器在block序列中遇到了新的dictionary时，它会使用这种启发式的方法确定是否继续推测。
在构建哈希表（如联接或聚合）时，Presto还会利用dictionary block结构。在处理索引时，operator会在数组中记录每个dictionary 条目（entry）在哈希表中的位置。如果有条目在后续的索引中重复，他会简单的重用该条目的的位置而不是重新对他进行计算。当连续的blocks共享同一dictionary 时，page处理器将保留该数组，来进一步减少必要的计算。
Presto在执行期间还会产生中间压缩的结果。例如，连接处理器（join processor）会在效率更高的情况下生成dictionary或RLE block。对于哈希联接，当联接的查找侧在哈希表中查找键时，它将值索引记录到数组中，而不是复制实际数据。Operator只是简单地生成了一个dictionary block，其中索引列表是该数组，而dictionary是对哈希表中该block的引用

Ⅵ. 性能

在这一部分，我们提供能了性能测试的结果，展现了本文所描述的设计决策对性能的影响。

A. 适应性

在Facebook内，我们在生产环境中运行着几种不同的connectors，来允许用户处理存储在各种内部系统中的数据。 表1描述了第II节中提到的几种案例的中的connector和部署方式。为了演示Presto是如何适配connector的特性，我们在30TB的数据上进行TPC-DS基准测试来对比查询时间。尽管Presto能够运行所有的TPS-DS查询，但是为了本次实验，我们选择了那些不需要溢出设置的查询子集。Presto: SQL on Everything（全文翻译） - 图6
表1:文中所述案例的部署情况
我们使用的是0.211版本的presto以及内部使用的类似于Hive/HDFS 和 Raptor的connector。Raptor是专为Presto设计的无共享存储引擎。它使用MySQL来存储元数据，并以ORC格式将数据存储在本地闪存盘上。Raptor支持复杂的数据组织方式（排序、分桶和临时列），但是对于本实验，我们的数据是随机分区的。Hive connector使用类似于Hive Metastore的内部服务，并在功能与HDFS相似的远程分布式文件系统上访问类似于ORC格式的文件。
每一个查询都会在如下三种配置的测试集群上运行：(1)Raptor中的数据存储在节点之间随机分布的表分片中。(2)Hive/HDFS中存储的数据，无统计信息。(3)Hive/HDFS中存储的数据带有表和列的统计信息。只要有这些统计信息，Presto优化器就可以根据成本决定连接顺序和连接策略。每个节点的配置为28核2.4GHz的Intel Xeon E5-2680 v4 CPU，1.6T闪存盘和256GB DDR4 RAM。
图6显示Presto在查询运行时受到connector特性的影响非常大。在不更改查询或群集配置的情况下，Presto能够通过利用connector的特性（包括吞吐量，延迟和可用的统计信息）来适应当前的connector。这也证明了Presto既可以作为传统的企业数仓也可以作为Hadoop数仓的查询引擎。Facebook的数据工程师经常会使用Presto对Hadoop数仓上的数据进行探索性分析，然后将汇总结果和经常访问的数据加载到Raptor中，以进行更快的分析和低延迟的仪表盘展示。Presto: SQL on Everything（全文翻译） - 图7
图6：执行部分TPC-DS查询的运行时间]

B. 灵活性

Presto的灵活性很大程度上归功于其低延迟的数据shuffle机制，以及支持对大量数据进行高性能处理的Connector API。图7显示了之前介绍的案例在我们生产环境中的查询运行时间的分布。我们只选取了那些执行成功且发生数据读取的查询。结果表明，Presto可以适用于具有严格延迟限制（20~100ms）的web服务，同样也适用于那些需要运行数小时的ETL作业。Presto: SQL on Everything（全文翻译） - 图8
图7：所述示例的查询时间分布

C. 资源管理

Presto集成了细粒度的资源管理系统，使它能够在不同的查询之间快速周转CPU和内存资源，以最大程度地提高在多租户群集中的资源利用效率。图8显示了来自Interactive Analytics案例集群的CPU和并发指标的4个小时的跟踪记录。即使从44个查询的峰值下降到8个查询的低点，Presto仍继续在每个Worker节点上平均使用约90％的CPU。同样值得注意的是，即便如此调度器仍然可以对那些低负载的查询进行优先执行并保持响应。一旦接收了新的查询，它会在毫秒级的时间内给这些查询分配大量的集群CPU资源。
Presto: SQL on Everything（全文翻译） - 图9
图8：群集4小时内的平均 的CPU利用率和并发性

Ⅶ. 工程相关

自2013年以来，Presto由Facebook的一个小团队开发，并对外提供服务。我们注意到，在快速发展的环境中，一些工程哲学对Presto的设计产生了巨大影响：
基于配置的自适应性：作为执行用户任意定义的计算的复杂的多租户查询引擎，Presto必须适应于不同的查询，还需要适有各种组合的特征。例如，在Presto能够自适应端到端的读写压力之前，部分具有慢客户端的作业会占用大量的内存和CPU资源，这会对同时正在运行的延迟敏感性的作业产生不利的影响。如果Presto不具备自适应性，则必须对工作负载进行分区，并分别调整每个工作负载的配置。但是这种方式无法扩展大我们生产环境中的各种查询用例中。
唾手可得的工具：Presto在查询（query）和节点（work node）层面提供了细粒度的性能统计信息。我们维护着自己的库（libraries），以便高效低地收集统计信息，而对于近似的数据结构我们使用扁平的内存进行存储。鼓励那些可被观察的系统设计并且允许工程师们检测和理解其代码的性能是非常重要的。我们的库使添加统计信息变得与注释方法一样容易。并且我们针对每一个查询收集了算子（operator）级别的统计信息，并将这些信息合并到task以及stage级别的统计信息中。我们在这监控上的投入，也使我们能够在优化系统时以数据为驱动。
静态配置：在像Presto这样的复杂系统中，操作相关的问题很难快速地找到根源并得到解决。属性配置会以难以理解的方式影响着系统的性能，因此我们优先考虑能够了解群集状态，而不是能够快速更改的配置。与Facebook的其他几个系统不同，Presto尽可能使用静态配置而不是动态配置。我们开发了自己的配置库，它被设计成在有任何警告的情况下，在启动时就会崩溃；这包括那些未使用，重复或冲突的属性。这种模式也面临着一系列的挑战。然而，对于大量的集群和配置项，将复杂性从操作上转移到部署过程或部署工具上会更有效。
垂直集成：与其他工程团队一样，我们为性能和效率等非常重要的组件设计了定制库。例如，自定义文件格式读取器允许我们端到端使用Presto原生的数据结构，避免了转换开销。然而，我们注意到，当操作一个多线程系统并在一个长生命周期的进程中执行任意计算时，轻松的调试和控制库行为的能力同样重要。
考虑一个近期生产问题的例子。Presto使用了Java内置的gzip库。在调试一系列的导致系统奔溃的操作时，我们发现glibc和gzip库（调用本地代码）之间的交互会导致内存碎片。 对于特定的工作负载组合，这会导致大量的内存泄漏。为了解决这个问题，我们改变了使用库的方式来影响正确的缓存刷新方式，但是在其他情况下，我们甚至为压缩格式编写了自己的库。
定制库还可以提高开发人员效率：仅实现必要的功能，统一配置管理并支持详细的工具以匹配我们的用例，从而减少了bug的产生。

Ⅷ. 相关工作

在过去十年中，针对大型数据集运行SQL的系统变得非常流行。而每一个系统都做了独特的取舍。对这些系统做全面的研究并不在本文的讨论范围之内。相反，我们将重点放在该领域的一些更值得注意的工作上。
Apache Hive最初是由Facebook开发的，为存储在HDFS中的数据提供一个类似sql的接口，并通过将它们编译成MapReduce或Tez作业来执行查询。Spark SQL是一个构建在流行的Spark引擎上的更现代的系统，它解决了MapReduce的许多限制。dSpark SQL可以在多个分布式数据存储上运行大型的查询，并且可以对内存中的中间结果进行操作。但是，这些系统不支持端到端管道，并且通常在stage间shuffle时将数据持久化到文件系统。虽然这提高了容错性，但是额外的延迟导致了这些系统不适合那些交互式或低延迟的用例。
Vertica ，Teradata，Redshift和Oracle Exadata等产品可以在不同程度上读取外部数据。但是，它们是基于内部的数据存储而构建的，当操作那些加载到系统内的数据时，能够达到最佳性能。一些系统采用了集成RDBMS风格和MapReduce执行的混合方式，如Microsoft SQL Server Polybase （用于非结构化数据）和Hadapt（为了性能）。Apache Impala可以提供交互式的延迟，但只能在Hadoop生态系统中运行。相反，Presto则与数据源无关。管理员可以使用垂直集成的数据存储(如Raptor)来部署Presto，但也可以配置Presto从各种系统查询数据（包括传统的关系型数据库以及NOSQL数据库）。Presto专有的内部服务和流处理系统，使得其在单个集群中也具有较低的资源开销。
系统和数据库社区在开发创新技术方面有着丰富历史，而Presto是在这些基础上开发的。它使用了类似于Neumann和Diaconu等人描述的技术，编制查询计划来加快查询处理速度。它可以使用Abadi等人的技术对压缩数据进行操作，并产生压缩的中间结果。可以从多个投影（projections ）中选择最理想的布局，并且使用了Zhou等类似的策略。通过推理计划属性来最大程度地减少shuffle。

Ⅸ. 总结

在本文中，我们介绍了Presto，这是一种facebook开发的开源的MPP SQL查询引擎，他可以快速的处理大型的数据集。Presto被设计成非常的灵活，可以将其配置成适用于各种用例的高性能的查询引擎。其丰富的插件接口和Connector API 使它具有很强的扩展性，从而可以和各种环境中的不同数据相结合。Presto同样被设计成自适应的，它可以利用connector提供的功能来加速执行，并可以自动调整读写并行度，网络IO，算子的启发方式并根据系统中运行的查询的特性进行调度。Presto的架构使其可以为低延迟的负载提供服务，并且能够有效的处理昂贵且长时间的查询。
Presto允许像Facebook这样的公司通过部署单个SQL系统处理多种常见的分析用例，来查询多个存储系统的数据，同时还支持扩展到上千个节点。它的架构设计使它能够在众多的大数据SQL引擎中占有一席之地。在Facebook以及整个行业中，Presto的使用率正在快速增长，并且我们的开源社区还在持续贡献着。