Facebook的数据库查询引擎Presto在美团的应用：配合Hadoop集群来优化大数据的处理

在少量大型Hadop/HDFS集群中存储Facebook数据仓库。Hive是Facebook几年前专门为Hadoop设计的数据仓库工具。过去，Facebook的科学家和分析师一直依靠Hive进行数据分析。但是Hive使用MapReduce作为底层计算框架，是专门为批处理设计的。然而，随着数据的增加，使用Hive进行简单的数据查询可能需要几分钟到几个小时，这显然不能满足交互式查询的需要。Facebook还研究了其他比Hive更快的工具，但要么功能有限，要么太简单，无法操作Facebook庞大的数据仓库。一些外部项目在2012年开始试用是不合适的，他们决定自己开发，这就是Presto。该项目于2012年秋季开始开发，目前已在1000多名Facebook员工中使用，运行了3万多个查询，每日数据为1PB级。Facebook称Presto的性能是Hive的10倍多。Faceboook于2013年正式宣布开源Presto。本文首先介绍了Presto从用户提交SQL到执行的过程，然后尝试分析总结Presto实现实时查询的原理，最后介绍了Presto在美团的使用情况。Presto架构Presto查询引擎是Master-Slave架构，由Coordinator节点、DiscoveryServer节点、多个Worker节点组成，DiscoveryServer通常嵌入Cordinator节点。Coordinator负责分析SQL语句，生成执行计划，将执行任务分发给Worker节点执行。Worker节点负责实际执行查询任务。Worker节点启动后，向DiscoveryServer服务注册，Cordinator从DiscoveryServer获得Worker节点，可以正常工作。如果配置了Hiveconector，则需要配置Hivemetastore服务，为Presto提供Hive元信息，Worker节点与HDFS互动读取数据。由于Presto是一个交互式查询引擎，我们最关心的是Presto实现低延迟查询的原理。我认为这主要是以下几个关键点，当然，还有一些传统的SQL优化原理，这里不介绍。为了介绍上述几点，首先介绍Presto执行查询的过程，提交查询用户使用PrestoCli提交查询句后，Cli使用HTTP协议与Cordinator通信，收到查询请求后，Coordinator调用SqlParser解析SQL语句获取Statement对象，并将Statement封装成QueryStarter对象，放入线程池中等待执行。像Hive一样，SQL编译过程Presto使用Antlr编写SQL语法，语法规则定义为Statementto.g和StatementBuilder.g两个文件中。如下图所示，从SQL编译到最终的物理执行计划大约分为五个部分，最终生成在每个Worker节点上运行的LocalexecutionPlan。这里不详细介绍SQL分析为逻辑执行计划的过程，通过SQL语句了解查询计划生成后的计算过程。样例SQL：复制代码如下：selectc11.rank,count(*)fromdim.cityc1joindim.cityc2onc1.id=c2.idwherec1.id>10groupbyc11.ranklimit100;上述SQL语句生成的逻辑执行计划Plan如上图所示。那么，Presto是如何将上述逻辑执行计划拆分，以更高的并行性来执行该计划的呢？让我们来看看物理执行计划。那么，Presto是如何将上述逻辑执行计划拆分并以更高的平行度执行该计划的呢？让我们来看看物理执行计划。物理执行计划逻辑执行计划图中的虚线是Presto对逻辑执行计划的切分点。逻辑计划Plan生成的Subplan分为四个部分，每个Subplan将提交到一个或多个Worker节点。Subplan有几个重要属性:plandistribution、outputPartitioning、partitionBy属性。Plandistribution表示查询Stage的分发方式。逻辑执行计划图中的四个Subplan有三种不同的Plandistribution方式：Surce表示Subplan是数据源，Source类型的任务将根据数据源大小确定分配多少个节点；Fixed表示，Subplan将分配固定的节点进行执行（Config配置中的query）.initial-hash-partitions参数配置，默认为8)；None表示，SubPlan只分配到一个节点进行执行。Subplan1和Subplan0Plandistribution在以下执行计划中=Source，这两个SubPlan都是提供数据源的节点，Subplan1所有节点的读取数据都会发送到Subplan0的每个节点；Subplan2分配8个节点进行最终聚合操作；Subplan3负责输出最终计算的数据。Outputpartioning属性只有两个值HASH和NONE，表示Subplan的输出是否根据partionby的key值对数据进行Shuffle。在下面的执行计划中，只有Subplan0的Outputpartioning=HASH，因此，Subplan2收到的数据是按照rank字段Partition后的数据。PrestoSQL的执行过程完全基于内存并行计算查询，如下图所示，Cli通过HTTP协议提交SQL查询后，查询请求包装成SqlQueryExecution对象交给cordinator的SqlQueryExecution线程池，执行每个SqlQueryExecution线程(图中Q-X线程)启动后，对查询请求的SQL进行语法分析和优化，最终生成多个Stage的SqlStagecution任务。每个SqlStageexecution任务仍然交给相同的线程池执行每个SqlStageexecution线程(图中S-X线程)启动后，每个Stage任务根据Plandistribution属性构建一个或多个Remotetask，通过HTTP协议分配给远端Worker节点执行Worker节点接收Remotetask请求后，启动Sqltaskexecution线程(图中T-X线程)，将此任务的每个Split包装成PrioritizedSplitrunner任务(图中SritioritizedSplitrner)-X）Taskexecutor#executor线程池交给Worker节点执行上述执行计划的实际执行效果如下图所示。通过HTTP协议调用Worker节点的/v1/task接口，Cordinator将执行计划分配给所有Worker节点(图中蓝色箭头)SubPlan1的每个节点读取一个Split数据，过滤后将数据分发给每个Subplan0节点进行Join操作和Partialagr操作。Subplan1的每个节点计算完成后，根据GroupbyKey的Hash值将数据分发给不同的Subplan2节点。计算完成后，将数据分发给Subplan3节点Subplan3节点计算完成后，通知Cordinator结束查询。Subplan1和Subplan0都是Source节点，并将数据发送到Cordinator源数据并行读取。事实上，他们读取HDFS文件数据的方式是调用HDFSInputSplitAPI，然后每个InputSplit分配一个Worker节点执行，每个Worker节点分配的InputSplit数量上限为参数可配置，Config中的query.max-pending-splits-per-默认为100的node参数配置。分布式Hash聚合的执行计划将在Subplan0中进行Partial聚合计算，计算每个Worker节点读取的部分数据的部分聚合结果，然后Subplan0的输出将根据Groupby字段的Hash值分配不同的计算节点，最后，Subplan3合并所有结果并输出流水线数据模型Presto中处理的最小数据单元是Page对象，如下图所示。一个Page对象包含多个Block对象，每个Block对象是一个字节数组，存储一个字段。一个Page对象包含多个Block对象，每个Block对象是一个字节数组，存储一个字段的几行。多个Block的横切线是真实的线数据。Page最大1MB，最多16*1024行数据。下图是Worker节点内部的计算流程图，左边是任务执行流程图。Worker节点将最细粒度的任务封装成PrioritizedSplitRunner对象，并将其放入pendingsplit优先队列中。每个Worker节点启动一定数量的线程计算，线程数task.shard.max-threads=availableProcessors()*4，配置在config中。每个免费线程从队列中取出一个PrioritizedSplitrunner对象执行。如果执行完成一个周期，超过最大执行时间1秒，则判断任务是否完成。如果完成，从allsplits队列中删除。如果没有，则将其放回pendingSplits队列中。每个任务的执行过程如下图右侧，依次遍历所有Operator，并尝试从最后一个Operator中获取一个Page对象。如果获得的Page不是空的，则交给下一个Operator执行。节点间流水线计算下图为Exchangeoperator执行流程图，Exchangeoperator为每个Split启动Httpagebuferclient对象，主动向上一个Stageworker节点拉数据，最小数据单位也是Page对象，在选择Source任务计算节点时，将数据放入Pages队列中本地化计算Presto，对于每个Split，根据以下策略，选择一些mincandidates，优先选择与Split相同的hostworker节点。如果节点不够，优先选择与Split相同的rackworker节点。如果节点不够，随机选择其他rack节点。对于所有candidate节点，选择assignedsplits最少的节点。动态编译执行计划Presto将ScanfilterandProjectoperator和FilterandProjectoperator动态编译成Bytecode，并将其交给JIT编译成native代码。Presto还使用了GoogleGuava提供的LoadingCache缓存生成的ByteCode。Presto还使用了由谷歌Guava提供的LoadingCache缓存生成的ByteCode。在上述两个代码片段中，第一段是动态编译前的代码，第二段是动态编译生成的ByteCode反编译后恢复的优化代码。我们可以看到，这里采用了循环优化方法。循环开发最常用于减少循环开支，并为具有多个功能单元的处理器提供指令级并行。也有利于指令装配线的调度。使用内存和数据结构使用Slice进行内存操作，Slice使用Unsafe#copyMemory实现高效的内存复制，Slice仓库参考：https://github.com/airlift/sliceFacebook工程师在另一篇介绍orcfile优化的文章中还提到，使用slice将orcfile的写作性能提高20%~30%，参考：https://code.facebook.com/posts/229861827208629/scaling-the-facebook-data-warehouse-to-300-pb为了加快avgg、countdistinct、聚合函数的查询速度，如percentile，Presto团队与BlinkDB作者之一Sameragarwal合作，引入了一些类似的查询函数aprox_avg、approx_distinct、approx_percentile。approx_distinct使用Hyperloglogllloglogllistinct