天穹SuperSQL是腾讯自研，基于统一的SQL语言模型，面向机器学习智能调优，提供虚拟化数据和开放式计算引擎的大数据智能融合平台。在开放融合的Data Cloud上，业务方可以消费完整的数据生命周期，从采集-存储-计算-分析-洞察。还能够满足位于不同数据中心、不同类型数据源的数据联合分析/即时查询的需求。
Presto在腾讯天穹SuperSQL大数据生态中，定位为实现秒级大数据计算的核心服务。主要面向即席查询、交互式分析等用户场景。Presto服务了腾讯内部的不同业务场景，包括微信支付、QQ、游戏等关键业务。日均处理数据量PB级，P90查询耗时为50s，全面提升各业务数据实时分析性能，有效助力业务增长。本篇文章将揭秘腾讯大数据在Presto上的核心工作，包括易用性、稳定性、性能，以及未来的主要方向等方面。
1 天穹Presto整体架构
天穹Presto作为天穹SuperSQL的主要执行引擎之一，主要承担着加速用户SQL执行速度的角色。SuperSQL通过智能路由（RBO/CBO/HBO）的方式，智能筛选出合适的SQL并分发给Presto执行，以加速SQL计算。另外，如果Presto执行失败，SuperSQL也能自动Failover到Hive或Spark重新执行，以确保用户SQL能顺利执行完成，而整个过程对用户透明、无感知。
关于天穹SuperSQL的介绍，可以查看历史文章：「解耦」方能「专注」——腾讯天穹SuperSQL跨引擎计算揭秘
天穹Presto采用了on K8s容器化部署的方式，具备自动化运维、弹性伸缩等云原生能力。同时支持为不同的业务独立部署专属的Presto集群，以避免造成不同集群负载的相互影响。
得益于Presto的多数据源访问能力，天穹Presto支持对接了Hive、Iceberg、MySQL等数据源，且为了更好地支持内部的业务，我们也扩展开发了内部的TDW Connnector，支持访问腾讯内部的数据仓库数据（TDW，Tencent distributed Data Warehouse）。同时天穹Presto使用了Alluxio作为数据源（Hive表、Iceberg表）的缓存层，用于加速热点数据的访问，可有效提升Presto查询的效率。
2 易用性增强
2.1 Hive语法兼容
由于部分用户习惯于使用Hive的语法，而Presto自身的语法语义与Hive相比又有些不同，因此天穹Presto在引擎侧做了兼容部分Hive语法语义的工作，主要包括：数值除法、数组下标取值、Hive UDF支持、Mapjoin Hint、隐式转换等。
对于Mapjoin Hint，其实是对应于Presto中的Broadcast Join，用户通过Mapjoin Hint来指定多表Join中的需要Broadcast（广播）的表，以此提升查询的性能，适用于大小表Join的场景。目前已支持在Inner Join和Left Join中使用Mapjoin Hint。
代码语言：javascript复制-- Presto采用Broadcast Join，Broadcast的表为test2
select t1.b, /* mapjoin(t2)*/ t2.b2 from test1 t1 join test2 t2 on t1.a = t2.a2;
-- Presto采用Broadcast Join，Broadcast的表为test1、test3
select t1.b, /* mapjoin(t1,t3)*/ t2.b2, t3,b3 from test1 t1 join test2 t2 on t1.a = t2.a2 join test3 t3 on t1.a=t3.a3;由于原生Presto不支持数值类型与字符串之间的隐式转换，为了兼容部分习惯于使用隐式转换的用户，天穹Presto在引擎侧做了增强，以支持类似于Hive语法中隐式转换的功能。
天穹Presto隐式转换规则表如下所示：（绿色表示支持从Source Type到Target Type的隐式转换，其余空白格表示不支持类型之间的隐式转换）
2.2 Query运行信息持久化
Presto的Event Listener提供了相关的接口，可以在查询执行完成后获取不同纬度的Metrics信息，比如查询执行期间各阶段的耗时、处理的数据量、内存/CPU消耗、Stage/Task/Operator统计信息等，天穹Presto扩展实现了Event Listener接口，将这些Query Metrics信息持久化到本地磁盘以及消息组件中，用于后续的问题定位、运维审计、资源统计、HBO等。
2.3 Iceberg Connector功能增强
腾讯天穹实时数仓-数据湖分析系统DLA使用了Iceberg作为表的数据组织格式，用户数据入湖后，可以通过Presto Iceberg Connector获得秒级的查询体验。天穹Presto也对Iceberg Connector做了一系列的功能增强，包括ORC存储格式支持（PR-16391）、Timestamp With Time Zone类型支持、Alluxio Local Cache支持（PR-16942）、并发写入（PR-16983）、Bugfix（PR-16959、PR-16968 ）等，大部分的特性或问题修复也已贡献到了PrestoDB社区。
3 稳定性提升
3.1 JVM调优
Presto在天穹上线运行的过程中，遇到过Worker Full GC停顿时间过长的问题，为此天穹Presto将JDK版本升级到了11（参考社区issue 14873），并对JVM参数做了持续的调优，比如适当增大-XX:GCLockerRetryAllocationCount参数的值（默认为2），以增加Full GC的概率尽量避免OOM的情况发生。目前在堆内存为180GB、CPU 96核的硬件条件下，天穹Presto Worker Full GC的平均耗时从数十秒降低到了十秒以内，停顿时间大幅下降。
3.2 Full GC Query Killer
Presto的查询内存使用统计是相对比较粗粒度的，这可能会导致原生的LowMemoryKillerPolicy在某些情况下不能正确地Kill查询，天穹Presto在线上运行的过程中就遇到过类似的情况：Worker堆内存已经接近用满了，但是Presto自身的Memory Pools显示还有较多的空闲内存，导致无法及时触发LowMemoryKillerPolicy。为了尽量避免这种情况，天穹Presto开发了Full GC Query Killer，该策略可以在Worker Full GC之后，如果Worker堆内存使用还是处于高值，则Kill掉在该Worker上使用最多内存的查询。需要注意的是，该策略是在应用程序层面执行的，如果Worker不断地Full GC乃至最后OOM，那么Full GC Query Killer可能也得不到响应，这时候还是需要通过其他手段分析定位出Full GC或OOM的原因，以彻底解决问题。
Full GC Query Killer相关的代码也将在近期贡献至PrestoDB社区，欢迎大家关注。
3.3 大文件ORC统计信息读取优化
Presto在读取ORC文件时，会先读取文件的Stripe统计信息，用于优化ORC的数据读取，但是如果ORC文件比较大，同时文件数量又比较多的情况下，StripeStatistics对象会占用较多的Worker堆内存，这些内存对象不断累积最终极易造成OOM。天穹Presto采用了以下的方案来尽量避免这个问题：对于来自同一个ORC大文件的Splits，避免重复读取文件的Stripe统计信息。
SplitFilerOperator会先读取一次ORC文件的Stripe统计信息，生成新的ORC Splits，新的Splits包含了利用Stripe统计信息过滤优化后的数据读取地址，后续ORC Splits分发至Worker中执行时，无需再读取Stripe统计信息，直接读取数据即可。
天穹内部环境测试结果显示该方案能减少50%左右的StripeStatistics对象内存占用，原先造成OOM的ORC查询，采用新方案的实现后也可以正常执行完成，目前正在上线生产环境中。
4 性能优化
4.1 Presto on Alluxio
天穹Presto on Alluxio主要有两种部署模式：Presto on Alluxio Cluster以及Presto Alluxio Local Cache，前者是比较通用的一种部署方式，但是需要额外维护一套Alluxio集群，Presto可以与Alluxio集群共部署或者分离部署，共部署的方式能有效提高本地读缓存的命中率，提升查询效率。Presto Alluxio Local Cache则是更轻量的部署模式，无需单独的Alluxio集群，数据缓存在Presto Worker侧，运维方便，缺点是Presto Worker动态扩缩容的场景下缓存会失效，目前PrestoDB和Alluxio社区也在持续推进Local Cache的方案，相信后续会越来越完善。
天穹Presto根据各业务的场景需求，对on Alluxio的方案做了以下增强，提高了易用性和可扩展性：
- 支持针对不同的Connector配置不同的Alluxio路由策略，比如Hive Connector和Iceberg Connector；
- 在Presto侧，新增Alluxio白名单机制，支持配置访问缓存在不同Alluxio集群下的库表数据；
- 在路由前检测Alluxio服务的状态可用性，当Alluxio服务不可用时自动Failover至HDFS；
白名单配置参数说明如下：
- "clusterUrl"：Alluxio集群的url地址，不同的集群可以配置不同的url。
- "tables"：Presto查询中涉及到的库表，如果已经在"tables"配置项中存在，则Presto会从对应的Alluxio集群中读取该库表的数据（首次从Alluxio中读取时，如果未有缓存，则Alluxio会将数据缓存下来，后续的读取会直接访问缓存），如果没有在"tables"中配置，则Presto会直接访问底层的HDFS、不经过Alluxio。"tables"支持库/表/分区级别的配置，支持通配符。
天穹Presto on Alluxio方案上线后，部分现网查询业务得到了20%～30%的性能提升，数据读取的耗时波动幅度变小、查询性能也更加稳定。
4.2 Presto on K8s
Presto on K8s是业界通用的一种部署模式，可以参考社区的presto-kubernetes-operator，天穹Presto根据自身的业务情况做了相应的适配改造， 整体的部署架构图如下所示：
每个Presto集群前端会部署一个CLB腾讯云负载均衡服务，对外提供统一的访问域名地址。CLB后端挂载Coordinator Pod，Worker通过CLB地址向Coordinator注册，客户端也通过CLB访问Presto。
天穹Presto集群有单独的租户资源，能保证集群的资源下限（Dedicated Resource），通过K8s HPA Controller感知Presto Worker的CPU和内存资源使用情况，实现Worker Pod的动态扩缩容。当白天Presto任务量较多需要更多资源时，可以动态扩容Worker至租户的资源上限，如果其他业务租户有空闲的资源，也可以继续"借用"。当晚上Presto集群空闲时，可以动态缩容Worker，将资源释放给其他业务租户使用，使资源池的利用率最大化。
4.3 Count Distinct Rewrite
Presto的Count Distinct实现在某些场景下会造成数据倾斜的问题，影响查询的性能，比如在Left Join之后再做Count Distinct，由于Presto use_mark_distinct规则的作用，会在Left Join之后做一次Repartitioning，然后在下一个Stage做MarkDistinct，如果Repartitioning阶段的Partition Key有较多重复值，那么就会造成下一个Stage出现数据倾斜的问题，影响MarkDistinct算子的执行速度。如果能将Count Distinct改写成Grouping Sets，由于Group By会在Repartitioning前做预聚合，所以能有效消除上述的数据倾斜问题。社区也有类似的issue 12024，但是从该issue的讨论内容来看，还并未有较完善的解决方案。
目前我们通过天穹SuperSQL来实现Count Distinct单列/多列到Grouping Sets的改写，无需改动Presto的代码，经过改写优化后，在某些用户场景下，能获得2～3倍的查询性能提升。
4.4 Optimized Repartitioning
天穹Presto每天的业务查询Exchange的数据量达到了上百PB级别，为了提升Repartitioning阶段的性能，我们在生产环境中启用了社区的Optimized Repartitioning特性（set session optimized_repartitioning=true; 参考 PR-13183），开启后，PartitionedOutputOperator算子整体的CPU消耗减少了50%，P90查询耗时降低了19%，某些用户场景下的查询性能提升接近2倍，节省了资源的同时性能也得到了较大的提升。
5 总结 & 未来工作
天穹SuperSQL的vision是通过构建大数据智能融合平台，将异构的计算引擎/异构的存储服务、计算的自动智能优化、流批一体的统一以及自治的系统运维纳入内部，给使用者提供简单统一的逻辑入口和虚拟化的视图方案，使得用户能够从繁杂的技术细节中解脱出来，专注于业务逻辑的实现。未来在Presto的工作主要有：语法扩展（临时表/视图的支持等）、运维增强（History Server、高可用）、自适应执行（运行在不同硬件规格的机器上）、内核性能提升、数据源Connector扩展增强等，在支撑好腾讯内部各业务需求的同时，也会积极拥抱和回馈开源社区，本篇文章的大部分内容，我们也在2021年12月举行的PrestoCon大会上做了分享PrestoCon-2021，欢迎大家持续关注。
5 联系我们
如果你对SuperSQL感兴趣，欢迎联系我们探讨技术。同时我们长期欢迎志同道合的大数据人才加入，欢迎咨询。联系方式：yikonchen@tencent.com
