Presto在大数据领域的实践和探索

我是谁？我从哪里来？要到哪里去？

Presto is an open source distributed SQL query engine for running interactive analytic queries against data sources of all sizes ranging from gigabytes to petabytes.
Presto allows querying data where it lives, including Hive, Cassandra, relational databases or even proprietary data stores. A single Presto query can combine data from multiple sources, allowing for analytics across your entire organization.
Presto is targeted at analysts who expect response times ranging from sub-second to minutes. Presto breaks the false choice between having fast analytics using an expensive commercial solution or using a slow "free" solution that requires excessive hardware.

• 清晰的架构，是一个能够独立运行的系统，不依赖于任何其他外部系统。例如调度，presto自身提供了对集群的监控，可以根据监控信息完成调度。

• 简单的数据结构，列式存储，逻辑行，大部分数据都可以轻易的转化成presto所需要的这种数据结构。

• 丰富的插件接口，完美对接外部存储系统，或者添加自定义的函数。

一览无余之Presto的架构

Presto的服务进程

Coordinator

• 从用户获得SQL语句

• 解析SQL语句

• 规划查询的执行计划

• 管理worker节点状态

Workers

一层一层剥开你的心之Presto数据模型

• catalog 对应某一类数据源，例如hive的数据，或mysql的数据

• schema 对应mysql中的数据库

• table 对应mysql中的表

• Page：多行数据的集合，包含多个列的数据，内部仅提供逻辑行，实际以列式存储。

• Block：一列数据，根据不同类型的数据，通常采取不同的编码方式，了解这些编码方式，有助于自己的存储系统对接presto。

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

• 完全基于内存的并行计算

• 流水线式的处理

• 本地化计算

• 动态编译执行计划

• 小心使用内存和数据结构

• 类BlinkDB的近似查询

• GC控制

欲先攻其事，必先利其器之Presto调优

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));

GENERAL_POOL is the memory pool used by the physical operators in a query.
SYSTEM_POOL is mostly used by the exchange buffers and readers/writers.
RESERVED_POOL is for running a large query when the general pool becomes full.

Query exceeded per-node total memory limit of xx
适当增加query.max-total-memory-per-node。

Query exceeded distributed user memory limit of xx
适当增加query.max-memory。

Could not communicate with the remote task. The node may have crashed or be under too much load
内存不够，导致节点crash，可以查看/var/log/message。

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

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

• Group By语句优化：合理安排Group by语句中字段顺序对性能有一定提升。将Group By语句中字段按照每个字段distinct数据多少进行降序排列， 减少GROUP BY语句后面的排序一句字段的数量能减少内存的使用.

• Order by时使用Limit， 尽量避免ORDER BY：Order by需要扫描数据到单个worker节点进行排序，导致单个worker需要大量内存

• 使用近似聚合函数：对于允许有少量误差的查询场景，使用这些函数对查询性能有大幅提升。比如使用approx_distinct() 函数比Count(distinct x)有大概2.3%的误差

• 用regexp_like代替多个like语句：Presto查询优化器没有对多个like语句进行优化，使用regexp_like对性能有较大提升

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

• 使用Rank函数代替row_number函数来获取Top N

• UNION ALL 代替 UNION ：不用去重

• 使用WITH语句：查询语句非常复杂或者有多层嵌套的子查询，请试着用WITH语句将子查询分离出来

它山之石可以攻玉之行业典型应用

• Hive SQL查询加速

• 数据平台Ad-Hoc查询

• 报表（BI报表、自定义报表）

• 活动营销

• 数据质量检测

• 资产管理

• 固定数据产品

• Hive SQL 兼容

• 物理资源隔离

• 直连Druid 的 Connector

• 多租户等

• 使用HDFS FileSystem Cache导致内存泄漏，解决方法禁止FileSystem Cache，后续Presto自己维护了FileSystem Cache

• Jetty导致堆外内存泄漏，原因是Gzip导致了堆外内存泄漏，升级Jetty版本解决

• Splits太多，无可用端口，TIME_WAIT太高，修改TCP参数解决

• Presto内核Bug，查询失败的SQL太多，导致Coordinator内存泄漏，社区已修复

• 通过Resource Group控制业务并发，防止严重超卖

• 通过JVM调优，解决一些常见内存问题，如Young GC Exhausted

• 善用MAT工具，发现内存瓶颈

• 数据平台(DP)的临时查询: 有赞的大数据团队使用临时查询进行探索性的数据分析的统一入口，同时也提供了脱敏，审计等功能。

• BI 报表引擎：为商家提供了各类分析型的报表。

• 元数据数据质量校验等：元数据系统会使用 Presto 进行数据质量校验。

• 数据产品：比如 CRM 数据分析，人群画像等会使用 Presto 进行计算。

• 第一阶段: Presto 和 Hadoop 混合部署

• 第二阶段: Presto 集群完全独立阶段

• 第三阶段: 低延时业务专用 Presto 集群阶段

• HDFS 小文件问题

node-scheduler.max-splits-per-node=100
node-scheduler.max-pending-splits-per-task=10

• 多个列 Distinct 的问题

   SELECT a1, a2,..., an, F1(b1), F2(b2), F3(b3), ...., Fm(bm), F1(distinct c1), ...., Fm(distinct cm) FROM Table GROUP BY a1, a2, ..., an

   转换为

   SELECT a1, a2,..., an, arbitrary(if(group = 0, f1)),...., arbitrary(if(group = 0, fm)), F(if(group = 1, c1)), ...., F(if(group = m, cm)) FROM
       SELECT a1, a2,..., an, F1(b1) as f1, F2(b2) as f2,...., Fm(bm) as fm, c1,..., cm group FROM
         SELECT a1, a2,..., an, b1, b2, ... ,bn, c1,..., cm FROM Table GROUP BY GROUPING SETS ((a1, a2,..., an, b1, b2, ... ,bn), (a1, a2,..., an, c1), ..., ((a1, a2,..., an, cm)))
       GROUP BY a1, a2,..., an, c1,..., cm group
   GROUP BY a1, a2,..., an

总结

参考列表：