Grouped Execution

为了方便大家理解 Grouped Execution 的原理，我们先来介绍两个概念：分桶 和 Hash Join。

1.1 分桶

• 其实 Hive 表中桶的概念就是 MapReduce 的分区的概念，两者完全相同。物理上每个桶就是目录里的一个文件，一个作业产生的桶（输出文件）数量和reduce任务个数相同。
• 而分区表的概念，则是新的概念。分区代表了数据的仓库，也就是文件夹目录。每个文件夹下面可以放不同的数据文件，通过文件夹可以查询里面存放的文件，但文件夹本身和数据的内容毫无关系。
• 桶则是按照数据内容的某个值进行分桶，把一个大文件散列称为一个个小文件。

1.2 Hash Join

主要分为两个阶段：建立阶段（build phase）和探测阶段（probe phase）

• Bulid Phase：选择一个表（一般情况下是较小的那个表，以减少建立哈希表的时间和空间），对其中每个元组上的连接属性（join attribute）采用哈希函数得到哈希值，从而建立一个哈希表。
• Probe Phase：对另一个表，扫描它的每一行并计算连接属性的哈希值，与bulid phase建立的哈希表对比，若有落在同一个 bucket 的，如果满足连接谓词（predicate）则连接成新的表。

1.3 Grouped Execution 原理

所谓的 Grouped Execution 针对的是那种数据摆放上进行了 bucketed 存储的数据查询的一种优化手段。

• 比如两个表 JOIN，在没有 Grouped Execution 的时候做 hash join，一般会把小表作为 build 表发到每个 worker 上，然后针对大表的数据做 probe，这样内存占用由整个小表数据量的大小决定

• 而如果这两个表是 bucketed 表，而且 bucket 的字段就是这个 JOIN 的字段，那么不同 bucket 之间的数据天然就 JOIN 不上，那么两个表之间的 JOIN 可以转变为相对应的 bucket 之间的 JOIN，最后再做个汇总即可，这样每个 worker 上的内存量会大大降低。

Recoverable Grouped Execution

单个 bucket 的 JOIN 如果失败是可以单独重试的，这也就引出了 Facebook 做的第二个优化：Lifespan 重试。通过 Lifespan 级别的重试可以提高大查询的成功率。

Exchange Materialization

上面两种优化手段的应用场景是很受限的，必须要 bucketed table，而且要用 bucketed 字段 JOIN 才能有用。确实，如果故事只到这里就结束就很没意思了，因此 Facebook 提出了第三个优化手段: Exchange Materialization。

我们知道 Presto 的 Exchange 本来是流式的，上游把数据通过 HTTP 发给下游，下游如果处理不过来会反压上游，中间是没有数据落盘的，如下图所示：

Exchange Materialization 则是要把数据落到盘上，并且按照 JOIN 的 key 组织成 bucketed table，那么从这个 Exchange 节点开始往后就可以应用上面的优化了，如下图所示：

Exchange Materialization 启用参数：

SET SESSION exchange_materialization_strategy='ALL';
SET SESSION partitioning_provider_catalog='hive';
SET SESSION hash_partition_count = 4096;

Spill to Disk

默认情况下，如果查询执行所请求的内存超过会话属性 query_max_memory 或 query_max_memory_per_node，Presto 就会终止查询。这种机制确保了查询分配内存的公平性，防止了内存分配导致的死锁。当集群中有很多小查询时，它是有效的，但会导致杀死不符合限制的大型查询。

4.1 原理简介

为了克服这种低效率，Presto 引入了可撤销内存的概念。Query 可以请求不计入限制的内存，但内存管理器可以在任何时候撤销这些内存。当内存被撤销时，查询运行程序将中间数据从内存溢出到磁盘，然后继续处理它。

在实践中，当集群空闲且所有内存都可用时，内存密集型查询可能会使用集群中的所有内存。另一方面，当集群没有太多空闲内存时，同一个查询可能被迫使用磁盘作为中间数据的存储。与完全在内存中运行的查询相比，强制溢出到磁盘的查询的执行时间可能要长几个数量级。

请注意，启用 spill-to-disk 并不能保证能够执行所有内存密集型的查询。查询运行程序仍然有可能无法将中间数据划分为足够小的块，使每个块都能装入内存，从而导致从磁盘加载数据时出现内存不足错误。

4.2 支持的操作

1、Joins

• 当任务并发大于 1 时，将对 build 表进行分区，分区的数量等于 task.concurrency 配置值
• 在对 build 表进行分区时，spill-to-disk 机制可以减少连接操作所需的峰值内存使用。当查询接近内存限制时，build 表分区的一个子集将溢出到磁盘，join 另一侧表中的记录也会落到相同的分区。
• 然后，就可以 one-by-one 读取溢出的分区数据以完成连接操作。
• 需要注意的是 join_spill_enabled 默认值是 false，可通过 set session join_spill_enabled=true; 启用

2、Aggregations

• 如果正在聚合的 group 中数量很大，则可能需要大量的内存。
• 当启用 spill-to-disk 时，如果没有足够的内存，则将中间累积聚合结果写入磁盘。
• 当内存可用时，它们被加载回来进行合并。
• 开启 spill_enabled 参数后无需单独设置其他参数

Dynamic Filtering

早在2005年，Oracle 数据库就支持比较丰富的 dynamic filtering 功能，而 Spark 和 Presto 在最近版本才开始支持这个功能。为了让大家快速理解这个功能，我们先来看一个 Spark 的栗子：

5.1 Spark 动态分区裁减

SELECT * FROM fact_iteblog
JOIN dim_iteblog
ON (dim_iteblog.partcol = fact_iteblog.partcol) 
WHERE dim_iteblog.othercol > 10

通过 Spark 的动态分区裁减，可以将执行计划修改成如下形式：

可见，在扫描 fact_iteblog 表时，如果能自动加上了类似于 select partcol from dim_iteblog WHERE dim_iteblog.othercol > 10 的过滤条件，那么当 fact_iteblog 表有大量的分区，而 select partcol from dim_iteblog WHERE dim_iteblog.othercol > 10 查询语句只返回少量的分区，这样可以大大提升查询性能。

5.2 Presto 动态过滤

Presto 也支持上面的功能，这个功能称为动态过滤（Dynamic Filtering）。事实上，Presto 的动态过滤比 Spark 的动态分区裁减要丰富。因为 Spark 动态分区裁减只能用于分区表，而 Presto 的动态过滤支持分区表和非分区表，Presto 的动态分区包含 Partition Pruning（分区表） 以及 Row filtering（非分区表）。直到 Presto 0.241，这个功能正式加入到 master 分支。

注意事项：

• 目前 PrestoDB 的实现中只有 Hive 数据源支持动态过滤
• 而且非分区表动态过滤只支持 ORC 数据格式
• 另外，Presto 的 dynamic filtering 和 grouped_execution 不能同时使用
• 并且需要设置以下参数

set session enable_dynamic_filtering=true;
set session hive.pushdown_filter_enabled=true;

Alluxio Cache Service

提高 Presto 查询延迟的一个常见优化是缓存工作集，以避免来自远程数据源或通过慢速网络的不必要的 I/O。Presto 利用 Alluxio 作为缓存层，主要有两种使用方式：Alluxio File System 和 Alluxio Structured Data Service。

注意：目前我们公司 Presto 集群尚未加入 Alluxio 缓存服务，这个优化手段可以作为一个知识储备。

6.1 Alluxio File System

Alluxio File System 将 Presto Hive Connector 作为一个独立的分布式缓存文件系统服务于 HDFS 或 AWS S3、GCP、Azure blob store 等对象存储之上。用户可以通过文件系统接口明确地了解缓存的使用情况并控制缓存。例如，可以预加载 Alluxio 目录中的所有文件，为 Presto 查询预热缓存，并设置缓存数据的 TTL（生存时间），以回收缓存容量。举个栗子：

注意对比 Hive 表的 location，前缀换成了 “alluxio://”

6.2 Alluxio Structured Data Service

Alluxio Structured Data Service 通过基于 Alluxio File System 的目录和缓存文件系统与 Presto 进行交互。这种方式有额外的优势，在不用修改 Hive 表的 location 的前提下，就能无缝访问现有的 Hive 表，并通过合并小文件或转换输入文件的格式进一步性能优化。具体做法如下：