Presto的介绍、使用和原理架构

简介

介绍

Presto是一款Facebook开源的MPP架构的OLAP查询引擎，可针对不同数据源执行大容量数据集的一款分布式SQL执行引擎，数据量支持GB到PB字节，主要用来处理秒级查询的场景。Presto 本身并不存储数据，，但是可以接入多种数据源，并且支持跨数据源的级联查询，而且基于内存运算，速度很快，实时性高。

注意：虽然Presto可以解析SQL，但它不是一个标准的数据库。不是MySQL、Oracle的代替品，也不能用来处理在线事务 (OLTP)。

适合：PB级海量数据复杂分析，交互式SQL查询，⽀持跨数据源进行数据查询和分析。 不像hive，只能从hdfs中读取数据。

不适合：多个大表的join操作，因为presto是基于内存的，多张大表在内存里可能放不下。

优缺点

优点：

• 清晰的架构，是一个能够独立运行的系统，不依赖于任何其他外部系统。例如调度，presto自身提供了对集群的监控，可以根据监控信息完成调度。
• Presto基于内存运算，减少了硬盘IO，计算更快．
• 能够连接多个数据源，跨数据源连表查，如从Hive查询大量网站访问记录，然后从Mysql中匹配出设备信息。
• 丰富的插件接口，完美对接外部存储系统，或者添加自定义的函数。

缺点：

Presto能够 处理PB级别的海量数据分析，但Presto并不是把PB级数据都放在内存中计算的。而是根据场景，如Count，AVG等聚合运算，是边读数据边计算，再清内存，再读数据再计算，这种耗的内存并不高．但是连表查，就可能产生大量的临时数据，因此速度会变慢。

Presto和hive的对比

hive是一个数据仓库（有hive表），是一个交互式比较弱的查询引擎，交互能力没有presto那么强，而且只能访问hdfs的数据(及数据源很单一)
presto是一个交互式查询引擎，可以在很短的时间内返回查询结果，秒级，分钟级，能访问很多数据源hive在查询100Gb级别的数据时，消耗时间已经是分钟级了

但是presto是取代不了hive的，因为presto全部的数据都是在内存中，限制了在内存中的数据集大小，比如多个大表的join，这些大表是不能完全放进内存的，所以presto不适合用在多个大表的join。

实际应用中，对于在presto的查询是有一定规定条件的：比如说一个查询在presto查询超过30分钟，那就kill掉吧，说明不适合在presto上使用，主要原因是，查询过大的话，会占用整个集群的资源，这会导致你后续的查询是没有资源的，这跟presto的设计理念是冲突的，就像是你进行一个查询，但是要等个5分钟才有资源给你用，这是很不合理的，交互式就变得弱了很多。我们理想的交互应该是实时的，速度越快越好。

Presto通过使用分布式查询，可以快速高效的完成海量数据的查询。如果你需要处理TB或者PB级别的数据，那么你可能更希望借助于Hadoop的HDFS来完成这些数据的处理。作为Hive和Pig（Hive和Pig都是通过MapReduce的管道流来完成HDFS数据的查询）的替代者，
Presto不仅可以访问HDFS，也可以操作不同的数据源，比如mysql。

Presto、Impala性能比较

https://blog.csdn.net/u012551524/article/details/79124532

测试结论：Impala性能稍领先于Presto，但是Presto在数据源支持上非常丰富，包括Hive、图数据库、传统关系型数据库、Redis等。

原理

架构

整体架构

Presto采用典型的master-slave模型，由一个Coordinator和多个Worker组成。

1. coordinator(master)负责meta管理，worker管理，query的解析和调度。Coordinator 跟踪每个 Work 的活动情况并协调查询语句的执行。Coordinator 为每个查询建立模型，模型包含多个Stage，每个Stage再转为Task 分发到不同的 Worker 上执行。Coordinator 与 Worker、Client 通信是通过 REST API。

2. Worker 是负责执行任务和处理数据。Worker 从 Connector 获取数据。Worker 之间会交换中间数据。最终结果会传递给 coordinator。Coordinator 是负责从 Worker 获取结果并返回最终结果给 Client。

   当 Worker 启动时，会广播自己去发现 Coordinator，并告知 Coordinator 它是可用，随时可以接受 Task。

   Worker 与 Coordinator、Worker 通信是通过 REST API。

3. discovery server， 通常内嵌于coordinator节点中，也可以单独部署，用于节点心跳，是将coordinator和work 结合到一起的服务，worker节点启动后向discovery server服务注册，coordinator从discovery server获得正常工作的worker节点。在下文中，默认discovery和coordinator共享一台机器。

4. Catelog。一个 Catelog 包含 Schema 和 Connector 。例如，你配置JMX 的 catelog，通过JXM Connector 访问 JXM 信息。当你执行一条 SQL 语句时，可以同时运行在多个 catelog。

   Presto 处理 table 时，是通过表的完全限定（fully-qualified）名来找到 catelog。例如， 一个表的权限定名是 hive.test_data.test，则 test 是表名，test_data 是 schema，hive 是 catelog。

   Catelog 的定义文件是在 Presto 的配置目录中。

5. Connector 是适配器，用于 Presto 和数据源（如 Hive、RDBMS）的连接。你可以认为 类似 JDBC 那样，但却是 Presto 的 SPI 的实现，使用标准的 API 来与不同的数据源交互。

   Presto 有几个内建 Connector：JMX 的 Connector、System Connector（用于访问内建的 System table）、Hive 的 Connector、TPCH（用于 TPC-H 基准数据）。还有很多第三方的 Connector，所以 Presto 可以访问不同数据源的数据。

   每个 Catalog 都有一个特定的 Connector。如果你使用 catelog 配置文件，你会发现每个 文件都必须包含 connector.name 属性，用于指定 catelog 管理器（创建特定的 Connector 使用）。 一个或多个 catelog 用同样的 connector 是访问同样的数据库。例如，你有两个 Hive 集群。 你可以在一个 Presto 集群上配置两个 catelog，两个 catelog 都是用 Hive Connector，从而达到可以查询两个 Hive 集群。

sql执行步骤

1. 客户端通过http发送一个查询语句给presto集群的coordinator
2. coordinator接收到客户端的查询语句，对语句进行解析，生成查询执行计划，并根据生成的执行计划生成stage和task，并将task分发到需要处理数据的worker上进行分析
3. worker执行task，task通过connector从数据源中读取需要的数据
4. 上游stage输出的结果给到下游stage作为输入，每个Stage的每个task在worker内存中进行计算和处理
5. client从提交查询后，就一直监听coordinator中的查询结果，一有结果就立即输出，直到轮训所有的结果都返回则本次查询结果结束

具体分析

（1）SQL 语句提交：

用户或应用通过 Presto 的 JDBC 接口或者 CLI 来提交 SQL 查询，提交的 SQL 最终传递给 Coordinator 进行下一步处理；

（2）词/语法分析：

首先会对接收到的查询语句进行词法分析和语法分析，形成一棵抽象语法树。然后，会通过分析抽象语法树来形成逻辑查询计划。

（3）生成逻辑计划：

下图是 TPC-H 测试基准中的一条 SQL 语句，表达的是两表连接同时带有分组聚合计算的例子，经过词法语法分析后，得到 AST，然后进一步分析得到如下的逻辑计划。

上图就是一棵逻辑计划树，每个节点代表一个物理或逻辑操作，每个节点的子节点作为该节点的输入。逻辑计划只是一个单纯描述 SQL 的执行逻辑，但是并不包括具体的执行信息，例如该操作是在单节点上执行还是可以在多节点并行执行，再例如什么时候需要进行数据的 shuffle 操作等。

（4）查询优化：

Coordinator 将一系列的优化策略（例如剪枝操作、谓词下推、条件下推等）应用于与逻辑计划的各个子计划，从而将逻辑计划转换成更加适合物理执行的结构，形成更加高效的执行策略。

下面具体来说说优化器在几个方面所做的工作：

• 自适应：Presto 的 Connector 可以通过 Data Layout API 提供数据的物理分布信息（例如数据的位置、分区、排序、分组以及索引等属性），如果一个表有多种不同的数据存储分布方式，Connector 也可以将所有的数据布局全部返回，这样 Presto 优化器就可以根据 query 的特点来选择最高效的数据分布来读取数据并进行处理。

• 谓词下推：谓词下推是一个应用非常普遍的优化方式，就是将一些条件或者列尽可能的下推到叶子结点，最终将这些交给数据源去执行，从而可以大大减少计算引擎和数据源之间的 I/O，提高效率。

  

• 节点间并⾏：不同 stage 之间的数据 shuffle 会带来很⼤的内存和 CPU 开销，因此，将 shuffle 数优化到最⼩是⼀个⾮常重要的⽬标。围绕这个⽬标，Presto 可以借助⼀下两类信息：

  • 数据布局信息：上⾯我们提到的数据物理分布信息同样可以⽤在这⾥以减少 shuffle数。例如，如果进⾏ join 连接的两个表的字段同属于分区字段，则可以将连接操作在在各个节点分别进⾏，从⽽可以⼤⼤减少数据的 shuffle。
  • 再⽐如两个表的连接键加了索引，可以考虑采⽤嵌套循环的连接策略。

• 节点内并⾏：优化器通过在节点内部使⽤多线程的⽅式来提⾼节点内对并⾏度，延迟更⼩且会⽐节点间并⾏效率更⾼。

  • 交互式分析：交互式查询的负载⼤部分是⼀次执⾏的短查询，查询负载⼀般不会经过优化，这就会导致数据倾斜的现象时有发⽣。典型的表现为少量的节点被分到了⼤量的数据。
  • 批量 ETL：这类的查询特点是任务会不加过滤的从叶⼦结点拉取⼤量的数据到上层节点进⾏转换操作，致使上层节点压⼒⾮常⼤。

针对以上两种场景遇到的问题，引擎可以通过多线程来运行单个操作符序列（或 pipeline），如图所示的，pipeline1 和 2 通过多线程并行执行来加速 build 端的 hash-join。

当然，除了上述列举的 Presto 优化器已经实现的优化策略，Presto 也正在积极探索 Cascades framework，相信未来优化器会得到进一步的改进。

资源和调度

查询调度

Presto 通过 Coordinator 将 stage 以 task 的形式分发到 worker 节点，coordinator 将 task 以 stage 为单位进行串联，通过将不同 stage 按照先后执行顺序串联成一棵执行树，确保数据流能够顺着 stage 进行流动。

Presto 引擎处理一条查询需要进行两套调度：

• 第一套是如何调度 stage 的执行顺序；
• 第二套是判断每个 stage 有多少需要调度的 task 以及每个 task 应该分发到哪个 worker 节点上进行处理。

（1）stage 调度

Presto 支持两种 stage 调度策略：All-at-once 和 Phased 两种。

• All-at-once 策略针对所有的 stage 进行统一调度，不管 stage 之间的数据流顺序，只要该 stage 里的 task 数据准备好了就可以进行处理；
• Phased 策略是需要以 stage 调度的有向图为依据按序执行，只要前序任务执行完毕才会开始后续任务的调度执行。例如一个 hash-join 操作，在 hash 表没有准备好之前，Presto 不会调度 left side 表。

（2）task 调度

在进行 task 调度的时候，调度器会首先区分 task 所在的 stage 是哪一类 stage：Leaf Stage 和 intermediate stage。Leaf Stage 负责通过 Connector 从数据源读取数据，intermediate stage 负责处理来此其他上游 stage 的中间结果；

• leaf stages：在分发 leaf stages 中的 task 到 worker 节点的时候需要考虑网络和 connector 的限制。例如蚕蛹 shared-nothing 部署的时候，worker 节点和存储是同地协作，这时候调度器就可以根据 connector data Layout API 来决定将 task 分发到哪些 worker 节点。资料表明在一个生产集群大部分的 CPU 消耗都是花费在了对从 connector 读取到的数据的解压缩、编码、过滤以及转换等操作上，因此对于此类操作，要尽可能的提高并行度，调动所有的 worker 节点来并行处理。
• intermediate stages：这里的 task 原则上可以被分发到任意的 worker 节点，但是 Presto 引擎仍然需要考虑每个 stage 的 task 数量，这也会取决于一些相关配置，当然，有时候引擎也可以在运行的时候动态改变 task 数。

split 调度

当 Leaf stage 中的一个 task 在一个工作节点开始执行的时候，它会收到一个或多个 split 分片，不同 connector 的 split 分片所包含的信息也不一样，最简单的比如一个分片会包含该分片 IP 以及该分片相对于整个文件的偏移量。对于 Redis 这类的键值数据库，一个分片可能包含表信息、键值格式以及要查询的主机列表。Leaf stage 中的 task 必须分配一个或多个 split 才能够运行，而 intermediate stage 中的 task 则不需要。

split 分配

当 task 任务分配到各个工作节点后，coordinator 就开始给每个 task 分配 split 了。Presto 引擎要求 Connector 将小批量的 split 以懒加载的方式分配给 task。这是一个非常好的特点，会有如下几个方面的优点：

• 解耦时间：将前期的 split 准备工作与实际的查询执行时间分开；
• 减少不必要的数据加载：有时候一个查询可能刚出结果但是还没完全查询完就被取消了，或者会通过一些 limit 条件限制查询到部分数据就结束了，这样的懒加载方式可以很好的避免过多加载数据；
• 维护 split 队列：工作节点会为分配到工作进程的 split 维护一个队列，Coordinator 会将新的 split 分配给具有最短队列的 task，Coordinator 分给最短的。
• 减少元数据维护：这种方式可以避免在查询的时候将所有元数据都维护在内存中，例如对于 Hive Connector 来讲，处理 Hive 查询的时候可能会产生百万级的 split，这样就很容易把 Coordinator 的内存给打满。当然，这种方式也不是没有缺点，他的缺点是可能会导致难以准确估计和报告查询进度。

资源管理

Presto 适用于多租户部署的一个很重要的因素就是它完全整合了细粒度资源管理系统。一个单集群可以并发执行上百条查询以及最大化的利用 CPU、IO 和内存资源。

（1）CPU 调度

Presto 首要任务是优化所有集群的吞吐量，例如在处理数据是的 CPU 总利用量。本地（节点级别）调度又为低成本的计算任务的周转时间优化到更低，以及对于具有相似 CPU 需求的任务采取 CPU 公平调度策略。一个 task 的资源使用是这个线程下所有 split 的执行时间的累计，为了最小化协调时间，Presto 的 CPU 使用最小单位为 task 级别并且进行节点本地调度。

Presto 通过在每个节点并发调度任务来实现多租户，并且使用合作的多任务模型。任何一个 split 任务在一个运行线程中只能占中最大 1 秒钟时长，超时之后就要放弃该线程重新回到队列。如果该任务的缓冲区满了或者 OOM 了，即使还没有到达占用时间也会被切换至另一个任务，从而最大化 CPU 资源的利用。

当一个 split 离开了运行线程，Presto 需要去定哪一个 task（包含一个或多个 split）排在下一位运行。

Presto 通过合计每个 task 任务的总 CPU 使用时间，从而将他们分到五个不同等级的队列而不是仅仅通过提前预测一个新的查询所需的时间的方式。如果累积的 Cpu 使用时间越多，那么它的分层会越高。Presto 会为每一个曾分配一定的 CPU 总占用时间。

调度器也会自适应的处理一些情况，如果一个操作占用超时，调度器会记录他实际占用线程的时长，并且会临时减少它接下来的执行次数。这种方式有利于处理多种多样的查询类型。给一些低耗时的任务更高的优先级，这也符合低耗时任务往往期望尽快处理完成，而高耗时的任务对时间敏感性低的实际。

（2）内存管理

在像 Presto 这样的多租户系统中，内存是主要的资源管理挑战之一。

（1）内存池

在 Presto 中，内存被分成用户内存和系统内存，这两种内存被保存在内存池中。用户内存是指用户可以仅根据系统的基本知识或输入数据进行推理的内存使用情况(例如，聚合的内存使用与其基数成比例)。另一方面，系统内存是实现决策(例如 shuffle 缓冲区)的副产品，可能与查询和输入数据量无关。换句话说，用户内存是与任务运行有关的，我们可以通过自己的程序推算出来运行时会用到的内存，系统内存可能更多的是一些不可变的。

Presto 引擎对单独对用户内存和总的内存（用户+系统）进行不同的规则限制，如果一个查询超过了全局总内存或者单个节点内存限制，这个查询将会被杀掉。当一个节点的内存耗尽时，该查询的预留内存会因为任务停止而被阻塞。

有时候，集群的内存可能会因为数据倾斜等原因造成内存不能充分利用，那么 Presto 提供了两种机制来缓解这种问题–溢写和保留池。

2.溢写

当某一个节点内存用完的时候，引擎会启动内存回收程序，现将执行的任务序列进行升序排序，然后找到合适的 task 任务进行内存回收（也就是将状态进行溢写磁盘），知道有足够的内存来提供给任务序列的后一个请求。

3.预留池

如果集群的没有配置溢写策略，那么当一个节点内存用完或者没有可回收的内存的时候，预留内存机制就来解除集群阻塞了。这种策略下，查询内存池被进一步分成了两个池：普通池和预留池。这样当一个查询把普通池的内存资源用完之后，会得到所有节点的预留池内存资源的继续加持，这样这个查询的内存资源使用量就是普通池资源和预留池资源的加和。为了避免死锁，一个集群中同一时间只有一个查询可以使用预留池资源，其他的任务的预留池资源申请会被阻塞。这在某种情况下是优点浪费，集群可以考虑配置一下去杀死这个查询而不是阻塞大部分节点。

内存管理

Presto是一款内存计算型的引擎，所以对于内存管理必须做到精细，才能保证query有序、顺利的执行，部分发生饿死、死锁等情况。

内存池

Presto采用逻辑的内存池，来管理不同类型的内存需求。

Presto把整个内存划分成三个内存池，分别是System Pool ,Reserved Pool, General Pool。

1. System Pool 是用来保留给系统使用的，默认为40%的内存空间留给系统使用。系统内存是实现决策(例如 shuffle 缓冲区)的副产品，可能与查询和输入数据量无关。

2. Reserved Pool和General Pool 是用来分配query运行时内存的。

   其中大部分的query使用general Pool。 而最大的一个query，使用Reserved Pool， 所以Reserved Pool的空间等同于一个query在一个机器上运行使用的最大空间大小，默认是10%的空间。

   General则享有除了System Pool和General Pool之外的其他内存空间。

换句话说，用户内存是与任务运行有关的，我们可以通过自己的程序推算出来运行时会用到的内存，系统内存可能更多的是一些不可变的。

为什么要使用内存池

System Pool用于系统使用的内存，例如机器之间传递数据，在内存中会维护buffer，这部分内存挂载system名下。

那么，为什么需要保留区内存呢？并且保留区内存正好等于query在机器上使用的最大内存？

Presto 引擎对单独对用户内存和总的内存（用户+系统）进行不同的规则限制，如果一个查询超过了全局总内存或者单个节点内存限制，这个查询将会被杀掉。当一个节点的内存耗尽时，该查询的预留内存会因为任务停止而被阻塞。

有时候，集群的内存可能会因为数据倾斜等原因造成内存不能充分利用，那么 Presto 提供了两种机制来缓解这种问题–溢写和保留池。

• 溢写：当某一个节点内存用完的时候，引擎会启动内存回收程序，现将执行的任务序列进行升序排序，然后找到合适的 task 任务进行内存回收（也就是将状态进行溢写磁盘），直到有足够的内存来提供给任务序列的后一个请求。

• 如果集群的没有配置溢写策略，那么当一个节点内存用完或者没有可回收的内存的时候，预留内存机制就来解除集群阻塞了。

  这种策略下，查询内存池被进一步分成了两个池：普通池和预留池。

  当query非常多，并且把内存空间几乎快要占完的时候，某一个内存消耗比较大的query开始运行。但是这时候已经没有内存空间可供这个query运行了，这个query一直处于挂起状态，等待可用的内存。 但是其他的小内存query跑完后，又有新的小内存query加进来。由于小内存query占用内存小，很容易找到可用内存。 这种情况下，大内存query就一直挂起直到饿死。

  所以为了防止出现这种饿死的情况，必须预留出来一块空间，共大内存query运行。 预留的空间大小等于query允许使用的最大内存。

  Presto每秒钟，挑出来一个内存占用最大的query，允许它使用reserved pool，避免一直没有可用内存供该query运行。

  为了避免死锁，一个集群中同一时间只有一个查询可以使用预留池资源，其他的任务的预留池资源申请会被阻塞。这在某种情况下是优点浪费，集群可以考虑配置一下去杀死这个查询而不是阻塞大部分节点。

内存管理

Presto内存管理，分两部分：

• query内存管理

• • query划分成很多task， 每个task会有一个线程循环获取task的状态，包括task所用内存。汇总成query所用内存。
  • 如果query的汇总内存超过一定大小，则强制终止该query。

• 机器内存管理

• • coordinator有一个线程，定时的轮训每台机器，查看当前的机器内存状态。

当query内存和机器内存汇总之后，coordinator会挑选出一个内存使用最大的query，分配给Reserved Pool。

内存管理是由coordinator来管理的， coordinator每秒钟做一次判断，指定某个query在所有的机器上都能使用reserved 内存。那么问题来了，如果某台机器上，没有运行该query，那岂不是该机器预留的内存浪费了？为什么不在单台机器上挑出来一个最大的task执行。原因还是死锁，假如query，在其他机器上享有reserved内存，很快执行结束。但是在某一台机器上不是最大的task，一直得不到运行，导致该query无法结束。

数据模型

presto采取三层表结构：

1. catalog 对应某一类数据源，例如hive的数据，或mysql的数据
2. schema 对应mysql中的数据库
3. table 对应mysql中的表

presto的存储单元包括：

1. Page： 多行数据的集合，包含多个列的数据，内部仅提供逻辑行，实际以列式存储。
2. Block：一列数据，根据不同类型的数据，通常采取不同的编码方式，了解这些编码方式，有助于自己的存储系统对接presto。

不同类型的block：

• array类型block，应用于固定宽度的类型，例如int，long，double。block由两部分组成

  • boolean valueIsNull[]表示每一行是否有值。
  • T values[] 每一行的具体值。

• 可变宽度的block，应用于string类数据，由三部分信息组成

• • Slice ： 所有行的数据拼接起来的字符串。
  • int offsets[] :每一行数据的起始便宜位置。每一行的长度等于下一行的起始便宜减去当前行的起始便宜。
  • boolean valueIsNull[] 表示某一行是否有值。如果有某一行无值，那么这一行的便宜量等于上一行的偏移量。

• 固定宽度的string类型的block，所有行的数据拼接成一长串Slice，每一行的长度固定。

• 字典block：对于某些列，distinct值较少，适合使用字典保存。主要有两部分组成：

• • 字典，可以是任意一种类型的block(甚至可以嵌套一个字典block)，block中的每一行按照顺序排序编号。
  • int ids[] 表示每一行数据对应的value在字典中的编号。在查找时，首先找到某一行的id，然后到字典中获取真实的值。

核心问题之 Presto 为什么这么快？

我们在选择 Presto 时很大一个考量就是计算速度，因为一个类似 SparkSQL 的计算引擎如果没有速度和效率加持，那么很快就就会被抛弃。

美团的博客中给出了这个答案：

• 完全基于内存的并行计算
• 流水线式的处理
• 本地化计算
• 动态编译执行计划
• 小心使用内存和数据结构
• 类 BlinkDB 的近似查询
• GC 控制

和 Hive 这种需要调度生成计划且需要中间落盘的核心优势在于：Presto 是常驻任务，接受请求立即执行，全内存并行计算；Hive 需要用 yarn 做资源调度，接受查询需要先申请资源，启动进程，并且中间结果会经过磁盘。

Presto安装

Presto Server安装

（1）官网地址

https://prestodb.github.io/

（2）下载地址

https://repo1.maven.org/maven2/com/facebook/presto/presto-server/0.196/presto-server-0.196.tar.gz

（3）将presto-server-0.196.tar.gz导入hadoop102的/opt/software目录下，并解压到/opt/module目录

[atguigu@hadoop102 software]$ tar -zxvf presto-server-0.196.tar.gz -C /opt/module/

（4）修改名称为presto

[atguigu@hadoop102 module]$ mv presto-server-0.196/ presto

（5）进入到/opt/module/presto目录，并创建存储数据文件夹

[atguigu@hadoop102 presto]$ mkdir data

（6）进入到/opt/module/presto目录，并创建存储配置文件文件夹

[atguigu@hadoop102 presto]$ mkdir etc

（7）配置在/opt/module/presto/etc目录下添加jvm.config配置文件

[atguigu@hadoop102 etc]$ vim jvm.config

添加如下内容

-server
-Xmx16G
-XX:+UseG1GC
-XX:G1HeapRegionSize=32M
-XX:+UseGCOverheadLimit
-XX:+ExplicitGCInvokesConcurrent
-XX:+HeapDumpOnOutOfMemoryError
-XX:+ExitOnOutOfMemoryError

（8）Presto可以支持多个数据源，在Presto里面叫catalog，这里我们配置支持Hive的数据源，配置一个Hive的catalog

[atguigu@hadoop102 etc]$ mkdir catalog
[atguigu@hadoop102 catalog]$ vim hive.properties 

添加如下内容

connector.name=hive-hadoop2
hive.metastore.uri=thrift://hadoop102:9083

如果是mysql，则创建一个mysql.properties：

connector.name=mysql
connection-url=jdbc:mysql://bd1:3306
connection-user=root
connection-password=

（9）将hadoop102上的presto分发到hadoop103、hadoop104

[atguigu@hadoop102 module]$ xsync presto

（10）分发之后，分别进入hadoop102、hadoop103、hadoop104三台主机的/opt/module/presto/etc的路径。配置node属性，node id每个节点都不一样。

[atguigu@hadoop102 etc]$vim node.properties
node.environment=production
node.id=ffffffff-ffff-ffff-ffff-ffffffffffff
node.data-dir=/opt/module/presto/data

[atguigu@hadoop103 etc]$vim node.properties
node.environment=production
node.id=ffffffff-ffff-ffff-ffff-fffffffffffe
node.data-dir=/opt/module/presto/data

[atguigu@hadoop104 etc]$vim node.properties
node.environment=production
node.id=ffffffff-ffff-ffff-ffff-fffffffffffd
node.data-dir=/opt/module/presto/data

（11）Presto是由一个coordinator节点和多个worker节点组成。在hadoop102上配置成coordinator，在hadoop103、hadoop104上配置为worker。

• hadoop102上配置coordinator节点

  [atguigu@hadoop102 etc]$ vim config.properties
  

  添加内容如下：

  coordinator=true
  node-scheduler.include-coordinator=false
  http-server.http.port=8881 //http端口号，presto主要用http请求
  query.max-memory=50GB
  discovery-server.enabled=true
  discovery.uri=http://hadoop102:8881
  

• hadoop103、hadoop104上配置worker节点

  [atguigu@hadoop103 etc]$ vim config.properties
  

  添加内容如下：

  coordinator=false
  http-server.http.port=8881  
  query.max-memory=50GB
  discovery.uri=http://hadoop102:8881
  

（12）在hadoop102的/opt/module/hive目录下，启动Hive Metastore，用atguigu角色

[atguigu@hadoop102 hive]$ nohup bin/hive --service metastore >/dev/null 2>&1 &

（13）分别在hadoop102、hadoop103、hadoop104上启动Presto Server

• 前台启动Presto，控制台显示日志

  [atguigu@hadoop102 presto]$ bin/launcher run
  [atguigu@hadoop103 presto]$ bin/launcher run
  [atguigu@hadoop104 presto]$ bin/launcher run
  

• 后台启动Presto

  [atguigu@hadoop102 presto]$ bin/launcher start
  [atguigu@hadoop103 presto]$ bin/launcher start
  [atguigu@hadoop104 presto]$ bin/launcher start
  

（13）日志查看路径/opt/module/presto/data/var/log

Presto命令行Client安装

（1）下载Presto的客户端

https://repo1.maven.org/maven2/com/facebook/presto/presto-cli/0.196/presto-cli-0.196-executable.jar

（2）将presto-cli-0.196-executable.jar上传到hadoop102的/opt/module/presto文件夹下

（3）修改文件名称

[atguigu@hadoop102 presto]$ mv presto-cli-0.196-executable.jar  prestocli

（4）增加执行权限

[atguigu@hadoop102 presto]$ chmod +x prestocli

（5）启动prestocli

[atguigu@hadoop102 presto]$ ./prestocli --server hadoop102:8881 --catalog hive --schema default

（6）Presto命令行操作

Presto的命令行操作，相当于Hive命令行操作。每个表必须要加上schema。

例如：

select * from schema.table limit 100

Presto可视化Client安装和使用

（1）将yanagishima-18.0.zip上传到hadoop102的/opt/module目录

（2）解压缩yanagishima

[atguigu@hadoop102 module]$ unzip yanagishima-18.0.zip
cd yanagishima-18.0

（3）进入到/opt/module/yanagishima-18.0/conf文件夹，编写yanagishima.properties配置

[atguigu@hadoop102 conf]$ vim yanagishima.properties

添加如下内容

jetty.port=7080
presto.datasources=atguigu-presto
presto.coordinator.server.atguigu-presto=http://hadoop102:8881
catalog.atguigu-presto=hive
schema.atguigu-presto=default
sql.query.engines=presto

（4）在/opt/module/yanagishima-18.0路径下启动yanagishima

[atguigu@hadoop102 yanagishima-18.0]$
nohup bin/yanagishima-start.sh >y.log 2>&1 &

（5）启动web页面

http://hadoop102:7080

看到界面，进行查询了。

（6）查看表结构

这里有个Tree View，可以查看所有表的结构，包括Schema、表、字段等。

每个表后面都有个复制键，点一下会复制完整的表名，然后再上面框里面输入sql语句，ctrl+enter键执行显示结果：

比如执行select * from hive.dw_weather.tmp_news_click limit 10，这个句子里Hive这个词可以删掉，是上面配置的Catalog：

常用SQL

我们把SQL分为几种类型来分别介绍：基础SQL、DDL(数据定义)、DML(数据操作)、DQL(数据查询)。基本上90%以上的ANSI SQL Presto都是支持的，使用起来也没有特殊的语法和限制。

基础SQL

查看系统中有哪些catalog，用SHOW CATALOGS;

查看指定catalog中有哪些schema，用 SHOW SCHEMAS FROM;

查看指定schema中有哪些table，用SHOW TABLES FROM;

如果需要查看指定表的建表结构，用 DESCRIBE（如果想查看建表语句，应该用SHOW CREATE TABLE）；

除此之外，还可以使用SHOW FUNCTIONS来查看系统中已经注册的Functions，用USE来切换catalog和schema。

以上的基础SQL，如show xxx，实际上在Presto的底层实现中，会把这样的SQL语句改变为select 语句来执行，如：

改变前的SQL：
show catalogs; 

改变后的SQL：
select * from (values ('catalog1'), ('catalog2'), (...)) as catalogs (catalog) order by catalog asc

DDL

Presto支持：

1. 创建表（Create Table）
2. 参照其他表来创建新表（Create Table Like）
3. 创建视图（Create View）
4. 创建表的同时插入数据（Create Table AS）

如：Create Table 与 Create Table Like：

// SQL1：
CREATE TABLE IF NOT EXISTS orders (
    orderkey bigint,
    orderstatus varchar,
    totalprice double,
    orderdate date
)
WITH (format = 'ORC');

// SQL2：创建表bigger_orders时，会根据此处定义的column和orders表中定义的column来创建
CREATE TABLE IF NOT EXISTS bigger_orders (
    another_orderkey bigint,
    LIKE orders INCLUDING PROPERTIES,
    another_orderdate date
);

关注一下，上面的SQL，使用到了WITH语法：

WITH ( property_name = expression [, ...] )

是否使用WITH是可选的，Presto使用它来在SQL中指定各种参数（property），执行这个SQL时，系统可以根据这些参数，产生不同的行为，相当于不改变SQL语法而扩展了SQL的表达能力。

一些传统的SQL On Hadoop技术，如Hive的做法是自行扩展了SQL的语法，而不是使用WITH，如下面SQL：

CREATE EXTERNAL TABLE page_views(
    view_time INT,
    user_id BIGINT,
    page_url STRING,
    view_date DATE,
    country STRING
) STORED AS ORC
LOCATION 'hdfs://user/hive/warehouse/analysis/';

如果用Presto SQL来表达相同的含义如下：

CREATE TABLE hive.analysis.page_views(
    view_time INT,
    user_id BIGINT,
    page_url STRING,
    view_date DATE,
    country STRING
) 
WITH (
    format = 'ORC',
    external_location = 'hdfs://user/hive/warehouse/analysis/';
)

可以看到Presto使用的是WITH方式，遵循了ANSI SQL标准，更加通用。WITH语法在Presto SQL中有着非常重要的作用，WITH中的的Properties，能表达很多丰富的语义，而且完全是可自定义的。这种WITH用法，与现在比较流行的FlinkSQL完全一样。

还有一种混合了DML和DDL的SQL，支持创建一张表，同时插入数据，它是**CREATE TABLE AS SELECT **，用起来也比较方便。如下SQL所示：

CREATE TABLE orders_column_aliased (order_date, total_price)
AS
SELECT orderdate, totalprice
FROM orders

需要注意的是，并不是所有的Connector都支持上面介绍的DDL，这个需要看特定Connector的实现，Presto的Connector执行框架为数据源提供了实现Create Table，Create View的机制，但是部分Connector没有实现，因为通过Presto来做Create Table和Create View的需求确实不多，除非是想用Presto做数据的ETL。当然，一些常用的Connector，如Hive Connector已经实现了，通过Presto来对Hive执行查询和数据读写操作也很方便。

DML

Presto 的Insert、Delete语法就是ANSI标准SQL语法，示例如下：

// SQL1：
INSERT INTO nation (nationkey, name, regionkey)
VALUES (26, 'POLAND', 3);

// SQL2：
INSERT INTO cities VALUES (2, 'San Jose'), (3, 'Oakland');

// SQL3：
DELETE FROM lineitem WHERE shipmode = 'AIR';

Presto 不支持Update语法，即使Connector指定为MySQL这种支持Update的数据源，也无法通过Presto来更新MySQL的数据。不过这并不是坏事，毕竟大数据OLAP系统中，核心操作是数据查询分析，数据更新需求几乎没有，而不是OLTP系统那样的CRUD操作。

需要注意的是，并不是所有的Connector都支持Insert，Delete，这个需要看特定Connector的实现，Presto的Connector执行框架为数据源提供了实现Insert，Delete的机制，但是部分Connector没有实现，因为通过Presto来做Insert和Delete的需求确实不多，除非是想用Presto做数据的ETL。当然，一些常用的Connector，如Hive Connector，已经实现了Insert和Delete，通过Presto来对Hive执行查询和数据读写操作也很方便。

DQL

Presto用户可以使用ANSI标准SQL来查询数据，下面给出了5个示例SQL：

// SQL1：查询表中的指定字段
SELECT name FROM tpch.sf1.region;

// SQL2：查询表中的指定字段，并且做一些条件过滤和排序
SELECT name FROM tpch.sf1.region WHERE name like 'A%' ORDER BY name DESC;

// SQL3：将两个表JOIN在一起输出，FROM后面跟两个表，用逗号连接，等同于A JOIN B这种形式。
SELECT nation.name AS nation, region.name AS region FROM tpch.sf1.region, tpch.sf1.nation WHERE region.regionkey = nation.regionkey AND region.name LIKE 'AFRICA';

// SQL4：用聚合函数(avg)计算给定字段的平均值，并向上取整。
SELECT round(avg(totalprice)) AS average_price FROM tpch.sf1.orders;

// SQL5：用SELECT子查询输出的结果来做条件过滤额
SELECT regionkey, name FROM tpch.tiny.nation WHERE regionkey =  (SELECT regionkey FROM tpch.tiny.region WHERE name = 'AMERICA');

插件

了解了presto的数据模型，就可以给presto编写插件，来对接自己的存储系统。presto提供了一套connector接口，从自定义存储中读取元数据，以及列存储数据。先看connector的基本概念：

1. ConnectorMetadata: 管理表的元数据，表的元数据，partition等信息。在处理请求时，需要获取元信息，以便确认读取的数据的位置。Presto会传入filter条件，以便减少读取的数据的范围。元信息可以从磁盘上读取，也可以缓存在内存中。
2. ConnectorSplit: 一个IO Task处理的数据的集合，是调度的单元。一个split可以对应一个partition，或多个partition。
3. SplitManager : 根据表的meta，构造split。
4. SlsPageSource : 根据split的信息以及要读取的列信息，从磁盘上读取0个或多个page，供计算引擎计算。

插件能够帮助开发者添加这些功能：

1. 对接自己的存储系统。
2. 添加自定义数据类型。
3. 添加自定义处理函数。
4. 自定义权限控制。
5. 自定义资源控制。
6. 添加query事件处理逻辑。

Presto提供了一个简单的connector : local file connector ,可用于参考如何实现自己的connector。不过local file connector中使用的遍历数据的单元是cursor,即一行数据，而不是一个page。 hive 的connector中实现了三种类型，parquet connector, orc connector, rc file connector。

Presto优化之数据存储

合理设置分区

与Hive类似，Presto会根据元数据信息读取分区数据，合理的分区能减少Presto数据读取量，提升查询性能。

使用列式存储ORC

Presto对ORC文件读取做了特定优化，因此在Hive中创建Presto使用的表时，建议采用ORC格式存储。相对于Parquet，Presto对ORC支持更好。

使用压缩Snappy

数据压缩可以减少节点间数据传输对IO带宽压力，对于即席查询需要快速解压，建议采用Snappy压缩。

内存调优

内存调优 Presto 有三种内存池，分别为 GENERAL_POOL、RESERVED_POOL、SYSTEM_POOL。

• GENERAL_POOL：用于普通查询的 physical operators。GENERAL_POOL 值为 总内存（Xmx 值）- 预留的（max-memory-per-node）- 系统的（0.4 * Xmx）。
• SYSTEM_POOL：系统预留内存，用于读写 buffer，worker 初始化以及执行任务必要的内存。大小由 config.properties 里的 resources.reserved-system-memory 指定。默认值为 JVM max memory * 0.4。
• RESERVED_POOL：大部分时间里是不参与计算的，只有当同时满足如下情形下，才会被使用，然后从所有查询里获取占用内存最大的那个查询，然后将该查询放到 RESERVED_POOL 里执行，同时注意 RESERVED_POOL 只能用于一个 Query。大小由 config.properties 里的 query.max-memory-per-node 指定，默认值为：JVM max memory * 0.1。

这三个内存池占用的内存大小是由下面算法进行分配的：

builder.put(RESERVED_POOL, new MemoryPool(RESERVED_POOL, config.getMaxQueryMemoryPerNode()));
builder.put(SYSTEM_POOL, new MemoryPool(SYSTEM_POOL, systemMemoryConfig.getReservedSystemMemory()));
long maxHeap = Runtime.getRuntime().maxMemory();
maxMemory = new DataSize(maxHeap - systemMemoryConfig.getReservedSystemMemory().toBytes(), BYTE);
DataSize generalPoolSize = new DataSize(Math.max(0, maxMemory.toBytes() - config.getMaxQueryMemoryPerNode().toBytes()), BYTE);
builder.put(GENERAL_POOL, new MemoryPool(GENERAL_POOL, generalPoolSize));

简单的说，RESERVED_POOL 大小由 config.properties 里的 query.max-memory-per-node 指定；SYSTEM_POOL 由 config.properties 里的 resources.reserved-system-memory 指定，如果不指定，默认值为 Runtime.getRuntime().maxMemory() 0.4，即 0.4 Xmx 值。而 GENERAL_POOL 值为：

总内存（Xmx 值）- 预留的（max-memory-per-node）- 系统的（0.4 * Xmx）。

Presto优化之查询SQL

只选择使用的字段

由于采用列式存储，选择需要的字段可加快字段的读取、减少数据量。避免采用*读取所有字段。

[GOOD]: SELECT time, user, host FROM tbl
[BAD]:  SELECT * FROM tbl

过滤条件必须加上分区字段

对于有分区的表，where语句中优先使用分区字段进行过滤。acct_day是分区字段，visit_time是具体访问时间。

[GOOD]: SELECT time, user, host FROM tbl where acct_day=20171101
[BAD]:  SELECT * FROM tbl where visit_time=20171101

Group By语句优化

合理安排Group by语句中字段顺序对性能有一定提升。将Group By语句中字段按照每个字段distinct数据多少进行降序排列。

[GOOD]: SELECT GROUP BY uid, gender
[BAD]:  SELECT GROUP BY gender, uid

Order by时使用Limit

Order by需要扫描数据到单个worker节点进行排序，导致单个worker需要大量内存。如果是查询Top N或者Bottom N，使用limit可减少排序计算和内存压力。

[GOOD]: SELECT * FROM tbl ORDER BY time LIMIT 100
[BAD]:  SELECT * FROM tbl ORDER BY time

使用Join语句时将大表放在左边

Presto中join的默认算法是broadcast join，即将join左边的表分割到多个worker，然后将join右边的表数据整个复制一份发送到每个worker进行计算。如果右边的表数据量太大，则可能会报内存溢出错误。

[GOOD] SELECT ... FROM large_table l join small_table s on l.id = s.id
[BAD] SELECT ... FROM small_table s join large_table l on l.id = s.id

注意事项

字段名引用

避免和关键字冲突：MySQL对字段加反引号**`、**Presto对字段加双引号分割

当然，如果字段名称不是关键字，可以不加这个双引号。

时间函数

对于Timestamp，需要进行比较的时候，需要添加Timestamp关键字，而MySQL中对Timestamp可以直接进行比较。

/*MySQL的写法*/
SELECT t FROM a WHERE t > '2017-01-01 00:00:00'; 

/*Presto中的写法*/
SELECT t FROM a WHERE t > timestamp '2017-01-01 00:00:00';

不支持INSERT OVERWRITE语法

Presto中不支持insert overwrite语法，只能先delete，然后insert into。

PARQUET格式

Presto目前支持Parquet格式，支持查询，但不支持insert。