Doris简史 - 为分析而生的11年
Apache Doris(incubating)从2008年第一个版本开始到今天已经走过了11个年头。期间,Doris 从最初的只为解决百度凤巢报表的专用系统,已经成长为目前国内唯一的分析型数据库孵化项目。一路走来, Doris 的初心从未改变。
Apache Doris —— 为分析而生
从诞生之日起,Doris 的每一步都是为了解决切实的业务痛点,每一次转变都是在面对不同的业务挑战。一路上,Doris 砥砺前行,凝结了众多前辈的心血。相信未来,Doris 还会有更多的新鲜血液加入,我们一起走的更快、更远。
Doris 发展历程
Doris 自第一版诞生以来,经过了11年的发展,中间做过无数改进。这只罗列对 Doris 发展来说比较重要的关键节点与事件。
#2008 Doris1,“筑巢引凤”的重要基石
早年,百度最主要的收入来源是广告。广告主需要通过报表服务来查看广告的展现、点击、消费等信息,并且能够需要通过不同维度来获得广告的消费情况,用以指导后续的广告的投放策略。
在 Doris1 诞生之前,百度使用 MySQL Sharding 方式来为广告主提供广告报表支持。随着百度本身流量的增加,广告流量也随之增加,已有的 MySQL Sharding 方案变得不再能够满足业务的需求。主要体现在以下几个方面:
第一,大规模数据导入会导致 MySQL 的读性能大幅降低,甚至还有锁表情况,在密集导入数据的情况下尤为明显。同时在数据导入时,MySQL 的查询性能大幅下降,导致页面打开很缓慢或者超时,用户体验很差;第二,MySQL 在大查询方面性能很差,因此只能从产品层面来限制用户的查询时间范围,用户体验很差;第三,MySQL 对数据量的支持是有限的。单表存储的数据有限,如果过大,查询就会变慢。对此的解决方案只有拆表、拆库、迁移数据。随着数据量的快速增长,已经无法维护。
当时数据存储和计算成熟的开源产品很少,Hbase 的导入性能只有大约2000条/秒,不能满足业务每小时新增的要求。而业务还在不断增长,来自业务的压力越来越大。在这种情况下,Doris1 诞生了,并且在2008年10月份跟随百度凤巢系统一起正式上线。
Doris1 的主要架构如上图所示。数据仍然通过用户 ID 进行 Hash,将同一个用户 ID 的数据交由一台机器处理。其中 Hm-Storage 负责数据的存储。ODP、OMG 负责将业务数据导入到 Hm-Storage 中。AS 负责解析、规划查询请求,并将查询请求发给 Hm-Storage 处理,并对 Hm-Storage 返回的数据进行一些业务相关的计算后将查询结果返回给用户。
相比于 MySQL 的方案,Doris1 主要在如下几个方面进行了改进。
首先,Doris1 的数据模型将数据分为 Key 列、Value 列。比如一条数据的 Key 列包括:用户 ID、时间、地域、来源等等,Value 列包括:展现次数、点击次数、消费额等。这样的数据模型下,所有 Key 列相同的数据 Value 列能够进行聚合,比如数据的时间维度最细粒度为小时,那同一小时多次导入的数据是能够被合并成一条的。这样对于同样的查询来说,Doris1 需要扫描的数据条目相比 MySQL 就会降低很多。
其次,Doris1 将 MySQL 逐条插入改成了批量更新,并且通过外围模块将同一批次数据进行排序以及预聚合。这样一个批次中相同 Key 的数据能够被预先聚合,另外排序后的数据能够在查询的时候起到聚集索引的作用,提升查询时候的性能。
最后,Doris1 提供了天表、月表这种类似物化视图的功能。比如用户是想将数据按天进行汇聚展现,那么对于这种查询是可以通过天表来满足的。而天表相对于小时表数据量会小几倍,相应的查询性能也会提升几倍。
通过 Doris1 的工作,完全解决了 MySQL Sharding 遇到的问题。并于2008年10月在凤巢系统一起上线,完美地支撑了广告统计报表需求。
#2009 Doris2,解“百度统计”燃眉之急
2008年的百度统计服务大约有50-60台 MySQL,但是业务每天有3000万+条增量数据,由于 MySQL 的存储和查询性能无法满足需求,对存量数据的支撑已经到了极限,问题频出,万般无奈之下百度统计甚至关闭了新增用户的功能,以减少数据量的增加。
Doris1 由于当时时间紧、任务重,所以设计、实现的时候只为了能够满足凤巢的业务需求,并没有兼顾其他的应用需求。由于 Doris1 方案对于凤巢成功的支持,百度统计同学开始基于 Doris1 打造 Doris2 系统,主要将 Doris1 进行通用化改造,包括支持自定义 schema 等,使 Doris 能够应用于其他产品。此外还进行一些优化以此来提升系统的查询、存储性能。
2009年 Doris2 研发完成后上线百度统计,并且成功支撑百度统计后续的快速增长,成功助力百度统计成为当时国内规模最大,性能、功能最强的统计平台。由于在凤巢、百度统计上的成功,公司内部后续其他类似统计报表类的需求也都由 Doris2 进行支持,比如网盟、联盟等报表服务。
#2010 Doris3 ,让查询再快一点
百度在2009-2011年发展迅猛,营收每年近100%的速度增长,与之相伴的是广告数据量也随之大幅增长。随着业务数据量的不断增长,Doris2 系统的问题也逐渐成为业务发展的瓶颈。
首先体现在 Doris2 无法满足业务的查询性能需求,主要是对于长时间跨度的查询请求、以及大客户的查询请求。这是因为 Doris2 通过规则将全部数据按照用户 ID 进行 Sharding,这虽然能够将全部数据分散到多台机器上,但是对于单一用户的数据还是全部落在一台机器上。随着单一用户数据量增多,一些查询请求无法快速计算得到结果。
其次,Doris2 在日常运维方面基本上都需要停服后手动操作,比如 Schema Change、集群扩缩容等,一方面用户体验很差,一方面还会增加集群运维的成本。最后,Doris2 本身并不是高可用系统,机器故障等问题还是会影响服务的稳定性,并且需要人肉进行复杂的操作来恢复服务。
为了解决 Doris2 的问题,团队开始了 Doris3 的设计、研发。Doris3 的主要架构如下图所示,其中 DT(Data Transfer)负责数据导入、DS(Data Seacher)模块负责数据查询、DM(Data Master)模块负责集群元数据管理,数据则存储在 Armor 分布式 Key-Value 引擎中。Doris3 依赖 ZooKeeper 存储元数据,从而其他模块依赖 ZooKeeper 做到了无状态,进而整个系统能够做到无故障单点。
在数据分布方面 Doris3 引入了分区的概念。首先数据会按照时间进行分区(比如天分区、月分区);在同一个分区里,数据会根据用户 ID 再进行 Sharding。这样同一个用户的数据会落在不同的分区上,而在查询时多台机器就能够同时处理一个用户的数据了,实现了单用户的分布式计算能力。但是可能还会存在一个分区内部单个用户数据量过大的情况。对于这种情况 Doris3 设计了后续表功能,会将单个分区内大用户的数据进行拆分,导入到多个分片中,这样能够保证每个分片内单个用户的数据总量最高是有限度的。
另外 Doris3 在日常运维 Schema Change,以及扩容、缩容等方面都做了针对性设计,使其能够自动化进行,不依赖线上人工操作。
在当时,由于种种原因,Doris3 最终确定使用了 Armor 来作为底层存储系统。Armor 是一款分布式 Key-Value 系统,支持多副本强一致,且单表内全 Key 有序。选用 Armor 作为底层存储能够使 Doris3 只负责管理分片,而分片的副本,以及副本的一致性都由 Armor 来处理。并且,集群的扩、缩容等操作也只需要 Armor 感知即可,Doris3 本身并不需要感知。当然除了这些好处外,这样的选型也有一些弊端。
由于 Armor 是一个通用的 Key-Value 系统,并不感知上层的业务数据,它并不支持 Doris 这种数据模型,相同 Key 的数据,Value 字段是可以进行聚合的。比如数据导入的批次是五分钟一批,但是数据时间粒度是小时,那么其实一个小时的数据可能是多次导入的,但是逻辑上是可以合并成一条数据的。所以为了实现这个功能,只能是 Doris3 自身实现了较为复杂的数据合并策略来完成相关数据的合并。
Doris3 在2011年完成开发后逐渐替换 Doris2 所制成的业务,并且成功解决了大客户查询的问题。而公司内部后续的新需求,也都由 Doris3 来承担支持。
#2012 MySQL + Doris3 ,百度的第一个 OLAP 平台
2012年随着 Doris3 逐步迁移 Doris2 的同时,大数据时代悄然到来。在公司内部,随着百度业务的发展,各个业务端需要更加灵活的方式来分析已有的数据。而此时的 Doris3 仍然只支持单表的统计分析查询,还不能够满足业务进行多维分析的需求。由于缺少通用的 SQL 支持,Doris3 在面对更加灵活的多维分析场景时有点力不从心。当时,公司内只有 Hive 以及类似系统支持大数据量的 SQL 查询,但是他们均是面向解决离线分析场景,而在线多维分析领域缺少一款产品来满足业务方的需求。
所以,为了能够支持业务的多维分析需求,Doris3 采用了 MySQL Storage Handler 的方式来扩展 Doris3。通过此种方式,将 Doris3 伪装成一个 MySQL 的存储后端,类似于 MyISAM、InnoDB 一样。这样既能够利用上 MySQL 对于 SQL 的支持,也能利用上 Doris3 对于大数据量的支持。由于这里 MySQL 是计算单点,为了减轻 MySQL 的计算压力,Doris3 应用了 MySQL 的 BKA(Batched Key Access)以及 MRR(Multi-Range Read)等机制尽量将计算下推到 Doris3 来完成,从而减轻 MySQL 的计算压力。
通过 MySQL + Doris3 这个方案,百度 Insight 团队为 PS、LBS、WISE 等产品线提供了百度内部第一个 OLAP 分析服务平台。
#2012 OLAP Engine,突破底层存储束缚
另一方面 Doris3 支持报表分析场景时,底层通用 Key-Value 存储引擎的弊端也逐渐显露。
第一,由于 Key-Value 系统读取只能够读取全 Key,全 Value,而报表分析系统中的大部分查询并不需要读取所有列,这样会带来不必要的 IO 开销;第二,正如前文所说,由于引擎本身不感知业务模型,不能够再进行 Merge 的同时完成数据的合并,这需要 Doris3 借助复杂的作业管理在引擎外部完成 Merge 工作既不简洁、也不高效;第三,为了保证业务的导入原子性,Doris3 为每批次导入都赋值一个版本号,并记录在每条数据 Key 的最后部分。这样在查询的时候,需要对每条数据进行 Key 的解析,比较版本号,过滤掉不需要的版本。这样一方面需要读取无需读取的数据,一方面需要解析所有 Key,从而带来不必要的 CPU 开销;第四,Key-Value 系统无法感知数据内容,只能使用通用压缩算法,进而导致数据的压缩效率不高。这样在查询、读取时都会带来较多的 IO 负载。
为了能够在底层存储引擎上有所突破,OLAP Engine项目启动了。这个项目的发起者是当时从 Google 来的高 T,为百度带来了当时业界最领先的底层报表引擎技术。OLAP Engine 最大的特点包括以下几点。
第一,引擎端原生就支持 Schema,并且所有的列分为 Key 列,Value 列。这样就能够跟上层的业务模型能够对应上,查询部分列时,无需加载全部列,减少不必要的 IO 开销。
第二,独特的数据模型。Value 列支持聚合操作,包括 SUM、MIN、MAX 等。在 Key 列相同的情况下,Value 列就能够按照聚合操作类型完成对应的聚合操作。而引擎本身导入方式类似于 LSM Tree,这样在引擎后台进行 Merge 的同时,就能够将相同 Key 的数据中的 Value 字段按照对应的操作进行聚合。这样就无需外部再进行数据合并作业管理,将引擎层与业务层合并合二为一,省去不必要的 IO、CPU 开销。
第三,数据批量导入,原子生效。对于每个批次的导入,都会有个 Delta 文件对应,并且会有个版本号。在查询的时候只是在初始化的时候来确定读取哪个文件,这样就只会读取生效版本的数据,而不会读取没有生效版本的数据,更不会浪费 CPU 来进行版本号比较过滤。
第四,行列式存储。多行(比如1024行)数据存储在一个 Block 内,Block 内相同列的数据一同压缩存放,这样可以根据数据特征利用不同的压缩算法(比如对于时间字段使用 RLE 等)大幅提高数据压缩效率。
即使分布式层没有采用复杂的分布式管理,只是使用类似 Doris2 的用户 ID Sharding 方式,OLAP Engine 后续也成功地支持了凤巢,网盟等广告业务。这充分体现了 OLAP Engine 强大的报表分析能力。虽然 OLAP Engine 取得了成功,但是由于硬 Sharding 方案带来的不易运维、不易扩展等问题仍然存在。
#2013 用 PALO,玩转 OLAP
底层技术的发展会激发上层业务的需求,而上层业务的需求同时会为底层的技术带来新的挑战。随着第一款 OLAP 产品的问世,数据分析师们的建模就更加复杂,有时查询 SQL 会有上千行,人为阅读已经相当吃力。而 MySQL + Doris3 方案的弊端也就越发突显。因为分析 SQL 越来越复杂,大量的计算都需要在 MySQL 中完成,这样 MySQL 的计算能力就成为整个系统的性能瓶颈,突破这个性能瓶颈也就变得极为紧迫。
因此 Doris 亟需一款拥有分布式计算能力的查询引擎。幸运的是当时(2013年)各种 SQL on Hadoop 项目也正蓬勃发展,比如 Impala,Tajo,Presto 等等。在有限的时间内并不充分调研的情况下,团队选取了 Impala 作为了后续系统的分布式查询引擎。当时选择 Impala 主要的原因是因为其性能较高,并且 BE 的 C++ 语言跟我们已有系统的语言一致,未来可以省去一部分序列化开销。
由于 MySQL + Doris3 的方案制约了业务的使用,当时公司的另一个团队邀请了 Oracle 的 Exadata 进行 POC,这给了 Doris 团队很大的压力。如果 Doris 想继续在 OLAP 领域继续发展,就需要快速产出原型,并且性能上还要胜出 Exadata。为了快速验证方案的可行性,团队几个月内就把 Impala 与 Doris3 进行了集成,并用 TPC-H 进行了测试,结果是 Impala + Doris3 性能比 Exadata 更好。这次原型的成功为我们赢得了一次机会,能够让团队继续改造 Doris3 从而更好地支持 OLAP 场景。
新产品的名字命名为 PALO,意为玩转 OLAP。
PALO1 除了增加分布式查询层之外,因为 OLAP Engine 在统计报表领域的成功,PALO1 放弃了 Doris3 依赖的通用 Key-Value 系统,选择了 OLAP Engine 作为自己的单机引擎。因为没有了分布式 Key-Value 系统,那么 PALO1 自己完成数据分片管理、副本管理等工作。
PALO1 的架构如下所示。其中 DM 负责管理元数据、数据的分布、分片副本管理等内容,DM 本身没有状态,元数据内容都存储在 MySQL 中。FE 负责接收用户的查询请求,并且进行查询规划解析。BE 是负责存储数据,以及进行具体的查询执行。
随着 PALO1 的正式上线,除了迁移所有 Doris3 已有的业务外,也成功支持了当时百度内部大部分的 OLAP 分析场景。
#2015 PALO2,让架构再简单一点
如果说 PALO1 是为了解决性能问题,那么 PALO2 主要是为了在架构上进行优化。由于 PALO1 模块数目较多,并且外部依赖 MySQL,这其实还是增加了运维的压力的。所以我们在 PALO2 项目中力求将系统的架构进行简化。经过简化后的系统架构如下图所示。
PALO2 中我们只存在2种模块:FE、BE。FE 一方面负责管理、存储元数据,另一方面 FE 还负责与用户交互,接受用户查询,对查询规划,监督查询执行,并将查询结果返回给用户。FE 本身是有状态的,但是它内部通过 BDB JE,能够将元数据进行多副本复制,从而能够保证服务的高可用。BE 与 PALO1 功能一致,只是 PALO2 的 BE 包含了存储引擎,一方面减少了一个模块,并且在用户查询的时候少了一次数据的序列化、反序列化操作,节约 CPU 消耗。
通过 PALO2 的工作,系统架构本身变得相当简洁,并且不需要任何依赖。因为 PALO2 架构的简洁,我们后续也相对容易的基于 PALO2 提供了公有云服务以及私有化部署;另一方面,当 PALO 开源之后其他用户也能够用通过较低的门槛来搭建使用 PALO 。在此之后 PALO 虽然经过几次改进,但是整体架构仍然保持 PALO2 的架构。
#2017 and Future Apache Doris (incubating) ,是更广阔的世界
PALO2 在百度内部基本服务了所有的统计报表、多维分析需求,我们相信它一定可以应用到其他公司,能够帮助更多的人更加高效、方便地支持类似的业务需求。因此,我们选择了开源,PALO 于2017年正式在 GitHub 上开源,并且在2018年贡献给 Apache 社区,并将名字改为 Apache Doris(incubating) 进行正式孵化。贡献给 Apache 之后,Doris 就不仅仅是百度的项目,而成为了 Apache 的项目。
随着开源,Doris 已经在京东、美团、搜狐、小米等公司的生产环境中正式使用,也有越来越多的Contributor 加入到 Doris 大家庭中。一路走来,Doris 从未惧怕过挑战,也从未被困难击倒。时至今日,Doris 已经站在了更高的舞台上,准备拥抱更多的机遇与挑战。
希望未来,会有更多的人来续写这篇 Doris 简史,讲述这个为分析而生的故事。