导读:火山引擎EMR作为一款云原生开源大数据平台产品,集成了包括Hadoop、Spark、Flink等引擎,并做到%开源兼容。ApacheDoris作为OLAP领域中一款极具代表性的开源组件,也被集成到了火山引擎EMR产品生态中。本文来源于火山引擎EMR团队大数据工程师在DorisSummit中的演讲实录,将为大家详细介绍火山引擎EMR是一款怎样的产品,火山引擎EMR团队对Doris社区做出了哪些贡献,火山引擎EMRDoris目前具备了哪些能力优化,以及后续的规划方向。
作者|字节跳动数据平台E-MapReduce团队昭伟
火山引擎是字节跳动旗下的云服务平台,将字节跳动快速发展过程中积累的增长方法、技术能力和工具开放给外部企业,提供云基础、视频与内容分发、数据平台VeDI、人工智能、开发与运维等服务,帮助企业在数字化升级中实现持续增长。
火山引擎EMR是一款云原生开源大数据平台产品。首先,从开源大数据平台角度,火山引擎EMR集成了开源大数据生态的众多软件栈,包括Hadoop、Spark、Flink等引擎,并且做到%开源兼容。ApacheDoris作为一款OLAP领域极具代表性的开源组件,所以我们也将其集成在火山引擎EMR生态中。其次,从云原生角度,我们也会基于云的特性做深度的能力增强,例如弹性伸缩、存算分离等。
我们在产品发布之初就已经集成了Doris引擎,它也是目前火山引擎EMR系统中的主力OLAP引擎之一。
EMRDoris是一个开箱即用的云端Doris服务。支持海量数据的高效导入、实时更新,支持对10PB级别的海量数据进行高并发查询。我们认为Doris也是一个比较全面的OLAP引擎,不同于ClickHouse仅在大宽表的聚合上较为擅长,Doris的各方面能力均比较出众。
支持向量化执行引擎,完备的MPP执行框架,多表关联能力更优异;
用户使用更友好,支持标准SQL以及MySQL协议;
支持预聚合表引擎,能方便快速地实现数据的聚合;
数据接入更便捷,提供多种数据导入方式以及异构数据源的访问能力;
具备物化视图的能力,能够实现查询自动路由,通过预计算来提高查询速度和并发。
因此,基于Doris全面的OLAP服务,火山引擎EMR很早对其进行了集成,进行了大量功能优化,并将其作为主力OLAP引擎之一推向了整个公有云市场。
基于Doris的能力优化
我们对Doris的研发主要以贡献开源社区为主,下面将为大家介绍下火山引擎EMR团队近期对于Doris社区的主要贡献。
基于Hudi的数据查询方案
在22年上半年,社区和火山引擎EMR团队一起做了基于Iceberg和Hudi的数据湖查询方案,火山引擎EMR团队主要负责Hudi的方案。
在此之前,查询Hudi需要通过一些导入工具,把Hudi的数据加载到Doris的内表中进行计算。但这样会带来一个坏处,数据链路相对来说会更长。因此我们把Doris作为一个数据服务层,和下层的Hudi做数据打通,实现数据的直接查询,避免数据反复导入导出。另外,通过打通HiveMetastore,来直接访问HiveMetastore获取库表的元数据,而不是通过表映射来关联字段,从而大大提升了数据开发的效率。通过以上优化,Doris实现无缝查询Hudi表。当然,目前这一方案只支持Hudi中CopyOnWrite(COW)存储类型的表,对MergeOnRead(MOR)表的支持尚在规划中。
MultiCatalog联邦查询
22年6~7月,我们和社区合作开启了MultiCatalog联邦查询的项目,目标是让Doris能像Presto一样有Plugin的能力,可以进行联邦查询,实现ES、JDBC等数据源的查询,以及最典型的Hive、数据湖的联邦查询。
最终我们通过Catalog直接查询Hive、Iceberg、Hudi表。经过了两个月的开发,目前已经支持三大数据组织模式,也支持数据存放在HDFS、S和TOS上,数据格式支持最常见的Parquet、ORC、TEXT等。基于这些能力,我们在性能上也做了持续的优化。例如,我们做了TableScan中最常见的几类优化,包括并发读取、RunTimeFilter、列裁剪、分区裁剪、Parquet和ORC中的谓词下推、数据预取等。做了这些有效的优化以后,相对于Trino,在同样的场景下,也就是Trino+HDFS或者Trino+S的模式,对比发现Doris的查询性能相比Trino要提升了近一倍左右。ComputeNode计算节点计算节点与联邦查询有很大的关联性。Doris本身是典型的Share-Nothing架构,所以在它的BE节点上计算和存储是强绑定的,这样会带来几个影响:扩容频繁,计算资源不够需要扩容,磁盘不够也需要扩容,只要满足一个条件,就必须要扩容。弹性能力差,因为每个节点都绑定了数据,一旦扩容就需要做数据的迁移。而一旦涉及到数据的迁移,时间相会比较长。而在联邦查询的场景下,用BE去承载联邦查询的计算相对来说比较厚重。基于这种假设,我们做了ComputeNode计算节点的功能。顾名思义,计算节点只有计算,没有存储,这样就非常适用于联邦查询场景。因为联邦查询本身没有数据存储,数据都是从远端拉过来的。另外它能很好地支持弹性,扩缩容的时候也不需要做数据搬迁,只要进程起来可以执行任务,资源过多时也可以在一两分钟之内将资源回收。因此计算节点可以实现非常好的弹性,可以支持分时弹性,这也是最常用的弹性策略。多表物化视图物化视图是一个典型的空间换时间的策略,通过预计算,配合查询时优化器的改写能力来直接查物化视图表,避免重复查询原表消耗过多的资源。Doris本身具有物化视图能力,但只针对单表的,可以对数据做简单聚合,因此我们也经常把它当做聚合索引。数据聚合后不再需要查原表,直接查物化视图性能会快很多,这也是它最大的使用场景。但该功能目前存在一些限制:仅支持简单聚合、不支持函数运算,例如当在Sum函数中嵌套加入CaseWhen语法时,物化视图无法创建。因此在即将发布的ApacheDoris2.0中,社区对单表物化视图进行了增强,可以支持更加复杂的SQL表达。Doris有比较好的MPP执行能力,经常会被用来做多表的关联查询,大宽表场景相对较少,因此如果物化视图只有单表时无法最大化发挥Doris的性能和场景优势。基于以上需求,我们联合社区一起研发了多表物化视图,并在即将到来的ApacheDoris2.0版本发布这一重大功能。届时可以将带有Join的查询结果固化以供用户直接查询,支持定时自动或手动触发的方式进行全量更新查询结果,后续也将进一步支持更加完善的自动增量刷新。
MySQLLoadDataLoadData是MySQL中的原生的语法,可以将数据从客户端加载到服务端的,几乎所有的MySQL生态都支持该语法,也是一种标准的数据导入方式。在Doris中可以导入本地数据的方式只有StreamLoad,它本身是Shell命令行,跟MySQL的语法有些差距,因此用户就需要做很大的改写。基于这种背景,我们希望能支持SQL语法导入客户端本地文件的能力。在具体实现中,我们按照MySQL网络协议,通过客户端包发送方式,把在本地的数据通过MySQL客户端直接发送到FE节点,再通过流式的方式封装成StreamLoadHTTPRequest发给BE,然后BE调用原来StreamLoad的逻辑把数据导入进去。这个功能实现MySQL生态的LoadData能力,也是Doris支持MySQL生态比较大的一个点。多流Upset多流Upsert源自于Flink中的多流Join,而多流Join需要维护比较大的状态,会导致集群不太稳定,因此很多OLAP引擎都支持部分列更新的能力、支持多流Upsert。我们基于原来的UniqueKey引擎实现了部分列更新的能力。具体能力如下图右侧所示,有两个Stream,它的主键是K1、K2,数据也有可能是乱序的,在要进行Merge的时候,会根据Key值把数据按照它的顺序合并起来。有该特性以后,数据的Join就变得比较简单,性能相对于原本使用Flink也有较大提升。节省了大量的回撤操作,所以在存储引擎去做这个能力还是非常不错的一种选择。我们有位客户需要对8表做Join的一个案例,因为客户之前是用Flink写到PostgreSQL中的,PostgreSQL本身支持多流Upsert,如果在Flink里面改写为join方式会非常困难。而Doris支持多流Upsert后,用法就跟PostgreSQL的用法完全一致,效果也会好很多。在性能方面,如果数据量不太大的时候,性能表现很优异,而当数据量特别大的时候,目前的这套实现还不是特别好,主要原因是多流Upsert是基于旧版本Merge-on-Read实现的,在读取时要做大量的合并操作。目前社区UniqueKey新的Merge-on-Write模式实现了性能的大幅优化,后续我们也将计划与社区一起进行全新UniqueKey部分列更新。
基于Doris的应用实践
以上是火山EMR对社区的贡献,下面将介绍我们基于Doris到底做出了一个怎样的产品。首先我们对客户提供的是云服务,所以除了引擎本身,引擎周边相对来说是我们重点布局的方面,例如下图右侧监控告警。如果是IDC用户,监控告警一般需要自己基于Grafana搭建。而作为云上的一款产品,提供完善的运维监控体系就可以大大简化用户的搭建成本。我们将相关的运维相关的功能直接在控制台页面透出,例如日志查询,以前日志需要自己去采集,现在也是完全不用户操心,直接在日志中心里面看到所有的服务器日志,用户只要在这个页面上查询,跟原来用ES去做搜集和用Kibana做展示的效果差不多的。除了监控、日志以外,还有集群的扩缩容能力,这也是云上的这种服务化能力优势的体现。对于集群节点配置,用户也不需要手动管理,而是直接使用配置中心,通过控制台页面很方便的进行配置更改。接下来的几个点,我们要谈谈火山引擎EMR相对于其他云上EMR产品,在OLAP引擎有哪些差异点。混合部署在下图场景中,FE和NameNode是在Master节点中混部的。实际上在计算的时候,只需要用到BE节点,但FE这三个节点又必须部署,不部署就无法正常工作。如果部署在BE中,又会受到隔离性的挑战。我们很多用户也有这种方面需求,希望能把FE节点和Master节点混部在一起。Hadoop的NameNode也与之类似,它不参与计算,只是作为一个RPC的响请求响应,做一些主备功能,或者做元数据管理等。由于不承载数据的大规模吞吐,所以NameNode和DorisFE的定位比较像。因此我们做了这样一个优化,将FE都部署在Master节点上,完成了一个混部。但计算节点目前我们不推荐用户去混部,因为Doris和Hadoop对磁盘、机型的要求不太一样的。Hadoop比较合适使用本地的HDD盘,但是Doris就是推荐本地SSD盘。所以我们做了这样的多Core节点组策略,也就是HadoopCore和DorisCore节点组,两者可以选择不同的机型和不同的磁盘类型,他们彼此独立的扩容节点,互不影响的情况下实现弹性能力。Master节点是混部的,随着业务增长Master节点也支持升配,实现垂直扩容能力。弹性节点组Doris本身不具备弹性的,而是有状态服务。但由于我们做了ComputerNode能力,也就使Doris支持了弹性能力。有了ComputerNode后,CN节点就可以按照要求就去做弹性策略了。我们可以完按照时间力度去做弹性,比如早上需要计算的数据多,就进行扩容,其他时间段,数据处理量不大就进行缩容。这也比较符合联邦查询的特点,一般都是白天负载会相对高一点。另外一个是基于集群负载的弹性能力实现自动的扩缩容。我们也支持基于时间和基于集群负载的弹性策略混合使用。自适应配置Doris目前的默认配置都是基于IDC的模式,一般都是按照大规格机器(G以上)去推荐设置的。这对于一些云上比较小的客户,比如0G以下的小规格的集群,会带来很多问题,我们之前遇到了很多OOM问题。因此我们基于不同的数据规模,设计了不同的集群规格,并针对每种规格地进行不同的配置优化。针对小规格,我们就将pagecache关闭,把bufferpool调小,并调低indexcache和Load内存配置,调小Session内存。针对大规格,我们主要是调大默认session内存和默认batch_size大小。中规格相对来说比较中庸,我们调小了pagecache,调低了loadpage内存配置和indexcache。通过这些配置的调整,我们完成了大中小规格的建设,减少用户的使用成本,也提高了整个集群的鲁棒性。未来规划
1.弹性能力,云产品最核心的价值是弹性能力,而Doris最初是基于IDC场景设计,弹性需求相对来说较低,所以短期内我们还需要对其进行大量的改造,来实现高弹性化。2.数据存储方面,因为Doris数据是自身进行管理,通过Tablet副本实现数据的高可用性。但在云时代,Doris仍旧自己管理数据其实没什么太大的必要,因为云上有S、TOS这些对象存储产品,它们能保证非常高的可用性,能达到了十几个9级别,这是通过Tablet副本很难实现的,这也是我们做存算分离的一个初衷。.存算分离方面,存算分离是把数据放到远端持久化存储中,通过缓存缓解一部分的查询性能压力。在此之上,我们整个集群就有了更好的弹性,剥离开数据之后的扩容也变的更加灵活方便,相当于做了一个stateless的BE,存算分离也能帮助用户进一步降低成本。另外在稳定性上也会有很大的提升,不需要自己通过Tablet副本实现高可用能力了,可用性可以直接交给S和TOS等对象存储产品来实现。整体系统复杂度也会大大降低,这也是存算分离能带来的一个非常大的优势。4.MetaServer,完成了数据的存算分离,数据弹性问题得到了解决,但元数据问题依旧存在。现在Doris的源数据是通过bdb做选主,通过本地做数据持久化,通过副本做高可用,这种模式依然依赖于节点来实现高可用。我们希望能做一个MetaServer,把元数据接到一个外部MetaServer中去,这样有几个好处:突破Master节点内存限制。现在的元数据全在内存中,当Tablet个数超过几千万的时候,内存消耗比较大。我们之前测过差不多万个Tablet的时候,内存达到将近20G,这20G的数据完全在内存里,没办法用磁盘去做溢出。如果有了MetaServer,比如基于MySQLRDS,那就只将一些热点缓存在FE中,其他的有需要的时候再去拉取,时延也和现在的模式不会有太大差别。
元数据的高可用性。MetaServer产品可以做到跨AZ级别,甚至跨region级别的高可用,但通过FE来说实现是非常困难的,这也是云上和云下的一个巨大的差异点,云上可以通过依赖标准的云产品的能力来实现自己能力,而在云下这些都需要自建。
节约成本。现在FE要通过三节点实现高可用,如果有了MetaServer,只要一节点就可以,成本也就随之降低。
支持更多类型的外部元数据存储,比如RDS,KV数据库等。
如果说前面提到的存算分离、MetaServer都是手段,那我们真正的目的是希望做到Doris的极致弹性化。
针对弹性化,集群规模可以跟随业务发展不断升级,这也是个云的基本能力。扩容有两个方向,一个scaleup,节点升降配都是比较常规的能力了。另外一个scaleout,这是本身Doris就有的,但是如果做了存量分离和MetaServer后,scaleout能力会进一步加强。有了这两个能力,我们才可以说Doris成为了一个真正意义上的云上Doris。