ITPub博客

F1 Query解析 | 前沿

原创 数据分析 作者:XDBTech 时间:2018-11-07 15:41:36 0 删除 编辑

Google最近在VLDB2018发表了"F1 Query: Declarative Querying at Scale"。SIGMOD2017上发表过一篇论文介绍Spanner的Query Engine。对比去年的遮遮掩掩,今年F1 Query诚意满满。X-DB的定位是一款对标Spanner的分布式HTAP数据库,作为X-DB的Query Engine的开发者,在研读后感慨良多,特此总结。

F1 Query是一个独立的SQL查询处理引擎,支持异构数据源的联邦查询。它从F1进化而来,早期只支持Spanner/Mesa,后来支持更多的异构数据源, 在设计和实现上向Dremel(数仓BigQuery的查询引擎)有所借鉴,但同时支持事务点查询。F1 Query是分布式的联邦查询处理引擎,即支持异构数据源。这是非常关键的一点,它影响了整个架构的设计。

F1 Query的定位

F1 Query能同时提供:

  • 低延时OLTP的点查询

  • 低延时的中等规模的OLAP查询

  • 大延时高吞吐的大规模的OLAP查询

可以看出,F1 Query定位于HTAP,提供企业级的各种数据处理和分析的需要。

F1 Query的架构

F1是基于存储计算分离的架构,底层的存储支持异构的数据源。F1 Query支持跨数据中心的部署,每个中心内都有一个完整的计算集群,包括:F1 Master,负责F1 servers的管理,以及所有查询的实时监控;F1 Server,接收到请求的节点,在分布式查询中也称为Coordinator节点,最终数据由此节点回Client;F1 Workers,一组参与计算的节点,与Server不一样的是,它仅是计算节点而不支持事务相关逻辑;F1 Catalog Service,目录服务器节点。其中,Master是数据中心内唯一此角色的节点,通过选举产生,不存在单点。

除了这些典型角色的节点外,还有一个负责大规模OLAP的处理模块,其中包括:F1 Distributor, 全局所有数据中心内唯一,同样是被选举产生,专门用以做分发;Batch Service, 用以调度、执行、资源管理,以及MapReduce的资源池。这个模块还有一些外部依赖,例如Spanner, 用于注册查询,以实现查询进度的监控及恢复。这些都是为了专门处理ETL的大规模OLAP需求。

总体看来,F1 Query的架构决定了许多特点,如可以在线线性扩展计算节点,可利用数据中心内的所有计算资源。但这种非data-locality的部署架构,会牺牲高性能低延时TP的能力。其内部又分了两套调度和执行框架,分别用于交互式和非交互式。所谓交互式,用户往往根据上条查询返回结果,再决定下条查询,用于典型的OLTP和时实或准时实的OLAP场景下。而非交互式,用于传统OLAP场景下。正是基于此,可以很好地处理非交互式的查询。其面向的问题场景,从TP到准实时的AP,到ETL处理的AP,但更偏向于AP,特别是异构数据源的处理。每个“全家桶”必然有所取舍,F1 Query在往高性能TP倾斜的场景下是短板。

优化器

F1 Query的优化器主要依赖于RBO(Rule-Based Optimization),采用Catalyst架构,应用规则集启发式地得到较优的计划。RBO中应该不包括一些高级Rule,比如基于Cost的改写等。CBO(Rule-Based Optimization)在文中没详细说明,应该是某种程度的使用,但提及这部分很难但正在开发中。除了RBO和CBO外,可以推断F1 Query不能很好地利用存储的统计信息做更精准的计划生成,原因是优化器所依赖的统计信息只针对部分特定数据源可用。这也是支持异构数据源的所带来的牺牲。

在逻辑计划的基础上,对其中算子确定数据源的Access Path和执行算法,生成物理计划。根据数据分布,排序,唯一性,估算的Cardinality等因素,来决定是生成Exchange算子,来桥接下层的输出和上层的输入,并按上层的需求,来对下层输出的数据做Reshuffle.

执行计划生成器,是优化器的最后一个阶段,生成可被调度的执行计划,并用DAG来表式,以供调度器所用。DAG本质上描述了Fragments的依赖关系。Fragment即为一组算子的集合。它有两个重要的因素:边界和并行度。优化器会自底向上,根据每个算子的输入的分布式需求,来计算每个Fragment的边界。当算子的输入的分布式需求并不需要跨多个Worker节点时,此算子可以与其下层的算子归为同一个Fragment,否则可能被划到不同的Fragment中。典型的算子包括JOIN/AGGR/SORT。每个Fragment内根据最底层SCAN的需求来设置并行度,上层的并行度则是下层的最大值。SCAN的并行度往往是分区的个数,并且会有最大上限,以防止分区数过大,从而会导致并行线程过多的问题。

总之,F1 Query在优化器上更多是依赖基于启发式规则的RBO,如条件下压,常量拆叠,投影裁剪,条件传播,Sort消除,公共子查询消除,物化视图的改写等。相比以AP为主的数据库系统,F1 Query在优化器上的工作还是偏少,但他们还在持续优化中。

调度器

F1 Query总地分为两种,交互式和非交互式。相应的,F1 Query的调度器由两个框架来实现,交互式由Coordinator来调度,非交互式则依赖于MapReduce的框架来调度。

对于交互式Query, F1 Query的调度器根据DAG来调度Query Plan. 调度Query Plan往往由请节发送到的节点,即Coordinator节点,来完成调度任务。根据Fragment内所描述的并行度N,在数据中心内,找N个负载较低的Worker节点,来参与执行。通过RPC往每个Worker上发送Fragment内执行计划,即所包括的算子集,以及必要的执行Context,驱动计划的执行。

除了单个Fragment内的并发外,各Stage的Fragment之间也会被调度起来,从而可以形成Pipeline,以加快整个执行的效率,除非有需要中断Pipeline的算子,例如SORT。前面提到,优化器会决定是否插入Exchange算子。在执行时,Exchange算子通过RPC,尽可能batch发送到下个Stage中。

对于非交互式,F1 Server生成Query Plan, 并向Spanner中注册Query。全局唯一的Distributor对在所有拥有此数据的数据中心中,根据负载和距离,来选择合适的数据中心,将Query分配给此中心。中心内的Query Scheduler定期地从Query Registry中捞Query来执行调度。同样地,它生成依赖图,对MP中的Worker池来执行Task.

小结下,对于交互式的分布式Query Plan, 调度器会同时调度各Stage的Fragments,对Fragment间的数据传输尽可能地Pipeline和Batch交互;而对于非交互式的Query Plan, 由全局唯一的Distributor来按Stage调度,并不会形成Pipeline。

查询执行

对于交互式和非交互式,同样在执行层面有两种实现。交互式即为典型的单机和分布式执行模式,支持TP及近似实时AP,非交互式则为一种特别的分布式执行模式,用以支持大型AP。正是不同的应用场景,决定了同是分布式查询,对应着不同执行策略,并且用不同的框架来实现。在很多细节上,像异常恢复,两者有着完全不同的实现策略。

交互式执行

如果一个查询中有2个分区的SCAN,是否一定生成分布式计划呢?不一样。如果优化器根据启发式规则估算出的SCAN数据量非常小,很可能为其生成单机的计划,以达到更好的性能。对于单机计划,则只需要一个F1 Server来处理,由单个线程完成计算,并将结果返回Client。类似的算子包括:SCAN, JOIN,  AGGR, ORDER等。单机执行就是传统的算子单机执行模式,没有什么特别之处,毕竟这一套都是成熟的实现方案。因为是计算存储分离,所以需要尽可能地尽少交换的数据量,例如Filter下压, 投影列裁剪, LIMIT和OFFSET下压等常见优化。

如果说单机计划下的DAG只有一个Fragment,那么分布式计划下的DAG则要复杂许多,除本身节点外,还会涉及其它多个Worker节点。接收到请求的节点被称之为Coordinator节点,它会根据优化器生成的DAG来做并发调度。在各Worker上执行时,各算子需要支持分布式。F1 Query支持典型的分布式算子,例如SCAN, JOIN,  AGGR, ORDER等。各Fragment之间数据消费的速度,对分布式的执行影响较大。例如,当消费者因为有锁依赖而无法消费下层发送过来的数据时,整修流程被阻塞。F1 Query的做法是,将内存中的数据写CFS。当然,这不是常态,因为常态下数据尽可能在缓存中,避免落盘。

分布式执行中,有两个因素对性能影响较大:Data Skew和次优化的Access Pattern。Data Skew在大数据处理中是比较常见的问题,典型的做法是调整Reshuffle的策略,包括调整并行度,优化分区函数,小表广播,  随机前后缀等。Acess Pattern的优化策略也比较多,如Batch Retrive等。F1 Query的优化策略也不出此外。例如,文中提及的Dynamic Range的策略,即在执行时,对分区的数据进行动态采样,以确定如何划分Range, 以尽量减少某个分片数据过多。

批量式执行

批量式执行依赖于MapReduce框架来实现。为了让Query Plan中的Fragments能映射到MapReduce的Stage, F1 Query将Map-Reduce的模型改两个Stage: Map-Reduce 和 Map<identify>-Reduce, 本质上是要将Exchange算子与MapReduce的Stage对接起来。与交互式下会将所有Fragment并发调度不同,MR并没有实现Pipeline, 而是严格按Stage来调度,在一个Stage全部完成后,将结果写入CFS,再开始调度下一个Stage。

批量式执行必须对所有Exchange Data做物化,以实现Task级别的恢复。对于特定的查询,会有一些优化的手段。例如,对于HashJoin, 如果输入的小表还是太大,或是关联较重,可以将小表物化成多个SST,从而将HashJoin转换成IndexJoin, 并且将IndexJoin算子与输入的大表算子归到同一个Fragment内,消除过多Exchange算子,从而实现减少中间物化的目的。

异构数据源的支持

依赖于Catalog Server获取数据存储的格式及系统,通过ProtoBuff作为数据Exchange的格式,最终在计算层将ProtoBuff的数据反弃列化后,转换成关系表中的列。计算层看到就是表的概念,从而实现计算层对异构数据源的统一支持。但文中提到,在处理不同的数据类型,如半结构化的XML等,遇到的挑战也不少。要处理好各种类型的数据,并且做到工业级强度,相信这里的工程量不会少。

容错机制

F1 Query有配套的F1 Client。当Client将请求路由到某数据中心后,F1 server根据Catalog Service检测到数据在其它数据中心时,直接给client报错,并附带数据所在的中心信息,以让client重新将查询路由到正确的数据中心。

除提供路由功能外,还提供Job级别的容错功能,即对Job的重试。但这仅仅适用于交互式查询,因为Job级别的只能依赖F1 Client来做重试。对于非交互式查询,则提供与Job更细粒度的Task级别重试。而这一套则依赖MapReduce来实现,在实现上,Exchange的中间结果物化到CFS中。

总结

F1 Query支撑了Google内部基础的数据查询服务,其技术在工程上得到了考验,其背后的技术,对阿里做数据库的同行,非常有借鉴意义。

Spanner去年在SIGMOD2017上发表过一篇论文介绍其Query Engine,对比今年在VLDB2018上分表的F1 Query, Spanner Query透露的信息量太少,而F1 Query则诚意满满。X-DB的定位是一款对标Spanner的分布式HTAP,那么X-DB和F1的Query Engine,其定位有什么区别,在技术上有哪些相同点,又有哪些不同点,我们后续会进行更为详尽地展述。

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/31556436/viewspace-2219091/,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2018-09-27

  • 博文量
    3
  • 访问量
    1040