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

以上就是关于Facebook的数据库查询引擎Presto在美团的应用：配合Hadoop集群来优化大数据的处理的相关介绍，更多Facebook的数据库查询引擎Presto在美团的应用：配合Hadoop集群来优化大数据的处理相关内容可以咨询我们或者浏览页面上的推荐内容。我们将让你对Facebook的数据库查询引擎Presto在美团的应用：配合Hadoop集群来优化大数据的处理有更深的了解和认识。