导读：火山引擎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。