货拉拉基于Doris的OLAP体系演进及建设方法

　　导读： 大家下午好，我是来自货拉拉大数据技术与产品部的杨秋吉，这次非常荣幸来到DataFun做分享。本次分享的内容是货拉拉基于Doris的OLAP体系演进及建设方法，希望能对大家有所帮助。
　　本次分享会分成五个部分：  背景介绍  OLAP体系演进（上）  OLAP体系演进（下）  总结思考和后续规划  QA环节
　　01
　　背景介绍
　　1. 总体介绍
　　货拉拉是成立于2013年，它成长于粤港澳大湾区，是一家从事同城、跨城货运、企业版物流服务、搬家、汽车销售及车后市场服务的互联网物流公司。截至2021年3月货拉拉的业务范围已经覆盖了就是国内352座城市，月活司机数也达到58万，月活用户达到760万，包含了8+个业务线。货拉拉大数据为了支撑公司业务，现在已经有了三个IDC，集群的机器数也达到了上千台，存储量达到了20PB+，日均任务数达到了20k+，并处在一个快速增长的一个过程中。
　　2. 大数据体系介绍
　　整个体系从下往上，可以看到分成了五层，从最下面的是一个 基础层和接入层 ，这两层主要是提供一个基础数据的一个存储、计算，还有一个集群的管理功能。在这之上是 平台层和数仓 。平台层包含了数据研发平台和数据治理平台。基于平台的一个能力和数据仓库的一个数据体系，在这之上包含了一些含有业务属性的 服务层和应用层 ，整个体系自下而上的话是相互支持的，最后实现了支持业务和赋能业务的一个能力。
　　3. 数据流介绍
　　这是一个比较典型的数据处理流流程 ，分成了一个 数据集成和采集 ，以及 数据的存储、计算和数据服务 这四部分，同时也包含了实时、离线、在线三大业务场景。
　　我们从左边看，在数据采集阶段会存在实时采集和离线采集两条路线。实时采集一个比较典型的场景是用户端上的埋点，它直接会同步到大数据里面做存储，供后续的在线和离线的计算使用。离线这块的数据主要是来自于业务方自己的数据库，会通过天或者小时定期的采集到大数据存储里以供后续使用。中间是数据的存储和计算阶段，离线场景是通过对数据的一个ETL以后转换成构造一个数仓的分层体系。实时这一块比较典型的一个场景是通过Flink做一个处理后，数据会直接落在线存储系统，类似于Hbase和OLAP等，供后续的业务服务业务数据。
　　--
　　02
　　OLAP体系演进（上）
　　货拉拉从2021年开始进行OLAP的一个技术研究，到现在已经经历了三个阶段。
　　1. 1.0孕育期
　　（1）业务场景
　　在1.0场景之前还没有引入OLAP之前业务的数据流。我们可以看到的是业务数据通过实时和离线处理以后，数据直接落到了MySQL，在MySQL里存的都是一些维度聚合以后的结果数据，意味着会在Flink里做很多的聚合分析。业务需要的相应维度的一系列组合，都会在Flink里面做实时的聚合，最后将结果存到MySQL。那么它带来的第一个问题是存在存储瓶颈，类似于KYLIN里的维度爆炸的问题。第二个问题是开发成本高、效率低。当业务方需要新增维度的时候，它需要对Flink里的所有作业都做一定的修改，然后再重新上线。第三个问题是部分的聚合需求支持不了。
　　（2）需求分析
　　基于这些存在的问题，我们分析其背后存在的需求，总结成了如下三点：
　　第一点是业务方希望能够横向扩容解决存储瓶颈问题；第二点是希望能够自由组合维度分析，提升他们业务开发效率；第三点是希望能够支持任意维度、时间跨度的分析。据此我们调研以后，决定采用OLAP引擎来支撑我们的业务需求。
　　业界的OLAP引擎多种多样，我们如何选择一款OLAP引擎并如何把它稳定的应用到生产？
　　（3）解决思路
　　首先是技术调研，技术调研需要充分理解业务需求，并结合业界的实践对比OLAP引擎的特点。在第二个阶段，选定选择好引擎之后，我们会做一系列POC验证，主要是通过功能、性能和数据质量这三个维度去做验证。验证通过以后我们再做大量的稳定性保障，主要是从服务的稳定性和数据链路稳定性做保障。在这之后就是业务上生产。
　　① 技术调研
　　我们对比了Druid、ClickHouse、Kylin、Presto、Doris等引擎。我们第一个业务诉求是希望存储这块能够横向扩展，基本所有的引擎都能够比较好的支持；其他诉求是希望可以自由组合维度分析，以及能够支持任意时间跨度的分析；除此之外，还一些隐含的业务需求，比如希望能够支持实时导入的精准去重和低延迟查询。
　　我们最终选择的是Druid，它除了能够满足我们的业务诉求之外，还有一个比较重要的影响原因是：因为它是纯Java开发，和我们的技术栈比较吻合，觉得可控性更高。
　　② POC验证
　　当我们选择好了一款引擎以后，我们会去做一个POC，主要会从这三个方面着手。  功能验证
　　从业务方的SQL查询模板把它提取出来，然后再根据Druid的Rollup语义做SQL改写，其中会涉及到大量的UDF改写以及语义兼容、Count Distinct函数等问题。  性能验证
　　直接采用真实的业务数据和业务SQL去执行，验证过程中我们把Cache关闭，分别统计了P75/P90/P99的查询耗时情况。在此期间我们发现有部分查询性能没有达到要求，那么剩下就是逐个分析。由于Druid没有一款比较完善的性能分析工具，不能够很好地打印出它的执行计划以及各个算子的耗时，所以我们采用了第三方的火焰图去做分析。
　　定位到相应的算子以后，最终通过一些优化将建表导出的逻辑和索引构建的逻辑，做了参数调整，主要是调整segment的大小并加入了物化视图去优化我们的性能。  数据准确性验证
　　把真实业务数据同时写Hive表和Druid，然后同时执行HiveSQL和DruidSQL去做数据质量的校验。在这个阶段过程中，我们发现一些函数如StringLast，会在一些特定的场景下会出现计算值不稳定的问题。
　　③ 稳定性保障
　　我们将整个稳定性保障过程拆成了事前、事中、事后三个阶段去做。
　　事前主要是做故障的预防，如容量规划、容灾演练，恢复预案；事中主要是锤炼自己的发现能力、定位能力，恢复和规避能力。发现能力主要就是构建一个全链路的监控，包括机器、服务、任务。事后主要是故障复盘、对应的一些整改措施以及如何落地。
　　④ 上生产
　　上线阶段我们同样分成了三个小阶段。
　　第一阶段是OLAP测试阶段，业务数据去会直接入到Druid，但业务的真实查询还是走原来的MySQL库。这个阶段我们主要是验证Druid的数据质量和集群稳定性。第二阶段是OLAP上线观察阶段，业务查询会切到Druid，但旧的MySQL链路没有下线，业务方能够随时切回MySQL链路，当OLAP运行稳定后，才会把MySQL的旧链路下线进行资源回收。
　　（4）问题总结
　　首先是 数据导入 这一块，一个比较典型的问题是实时数据乱序，它带来的影响的就是过多的小文件会影响查询效率且会增加Druid的元数据压力。解决方案是通过业务手段去规避，如在上游Flink里过滤一些异常数据。
　　第二个问题是 在数据准确性验证阶段发现StringLast的函数值不稳定 ，多次查询结果不一致。解决办法是新增加了一个StringLastMax和StringLastMin函数。
　　第三个问题是发现 Druid没有高效的精准去重函数，而业务场景又非常依赖于精准去重 。对于这种情况，我们引入了快手提供给社区的的一个Patch并把它合入到当时最新0.20版本,在这个基础之上,我们新增了一个SQL API并支持导入Hive的bitmap二进制字符串类型。
　　2. 2.0完善期
　　（1）业务需求分析
　　在2.0阶段，我们主要有如下一些 业务需求点 ：
　　第一是要 能够支持司机明细查询 ，第二是 能够支持多维的聚合分析 ，第三是 能够支持单天近10亿实时数据写入 ，因为是单表，所以需要比较高的写入吞吐。最重要的一点是需要能够支持复杂数据结构的高效写入查询，如Map和JSON格式。因为业务方是一个埋点数据，会有一些自定义的Map QA数据。
　　为了解决这个业务场景我们还是复用到1.0的解决思路，通过技术调研、POC稳定性上生产这四个步骤去实现。
　　① 技术调研
　　技术调研阶段，我们首先研究了Druid能不能很好地支持，发现并不是很理想，主要是Druid对于复杂的数据结构，支持度并不是很好，其次是Druid虽然能够支持明细查询，但它的明细和聚合是分成不同的表，这样无疑增加了存储成本。所以最终我们 又引入了ClickHouse组件 ，一是它能够比较好的支持复杂数据类型，二是它对于实时导入语义没有那么高的要求。
　　剩下的就是POC还有上生产这一系列的验证操作，这块和1.0比较类似，就不再赘述了。
　　--
　　03
　　OLAP体系演进（下）
　　经过了去年一整年的摸索，且随着公司业务的发展，更多的产品线对于多数据源关联场景的在线分析需求变得越来越迫切。比如说ABTest、实时数仓这些场景，对于多表关联需求，尤其是大表关联变得越来越迫切。
　　1. 需求分析
　　把AB实验的数据和后面相应的司机、用户埋点数据关联做分析。在这种情况下，会发现之前的两种组件都会存在一些弊端。为了解决这个问题，我们还是复用了1.0的解决思路，通过技术调研、POC稳定性和上生产去解决
　　2. 技术调研
　　在技术调研阶段，我们主要对比了Druid、ClickHouse和Doris。Druid支持维表的一些简单JOIN，ClickHouse能够支持类似BroadCast的内存JOIN，但针对千万级或者是亿级大数据量不是很友好，Doris能够支持小表的JOIN，对大表也支持基于Shuffle的JOIN。所以我们最终选择了Doris。
　　3. POC验证
　　在POC阶段，除了引用业务真实的数据和场景做验证以外，还引入了TPC-DS的数据集做了验证。在多表关联的场景下, 5亿+数据的JOIN TP75大概是9秒左右。在数据质量阶段，我们也是把TPC-DS的数据集，还有业务真实数据集，分别在Hive和Doris里面做了双跑验证，发现两者都是能够完全对得上的。
　　4. 稳定性保障
　　稳定性的保障方式还是和之前一样，变动的是监控手段，主要依托Compaction相关的一些监控。
　　5. 问题总结 问题1：查询性能优化
　　需求 ：查询7天数据，RT<=5s
　　优化前 ：查询7天数据耗时30s
　　问题2：UnhealthyTablet不下降， 查询报错-230
　　场景 ：不停flink写任务， be机器交替重启，重启完后出现unhealthyTablet。
　　原因:
　　① coordinator be在两阶段提交执行Commit后publish前被重启了。
　　② max_running_txn_num_per_db参数配置过大， compaction压力大。
　　解决办法:
　　① 引入社区1.10 patch（ issue-9267）
　　② 数据恢复
　　6. 参数优化
　　--
　　04
　　总结思考与后续规划
　　1. 总结思考
　　从业务需求去出发，然后去做一个匹配合适的一个引擎，为业务精细化运维提供一个技术支持。
　　摸索了一套比较完善的一个上线流程及稳定性保证方案，为业务的一个平稳运行提供能力保障。
　　很难有一个单个引擎能够富含谈各种的一个场景。所以的话在技术选型的时候，需要针对于需求特点和引擎的特点做一个比较合理的一个选择
　　2. 后续规划
　　希望去往OLAP平台化，去做一个自助化的一个建模。同时我们也会做一些多引擎的一个路由，在这一块让它能够支持一个明细和聚合的分析，还有关联的一个场景。除了平台化以后，我们的后续的一个引擎演进的话，是打算从高效稳定和内核演进这里面去做一个演进。关于内核演进，我们发现Doris基本是能够富含覆盖注意的所有的场景，所以我们后面计划着是以Doris引擎为主，Clickhouse引擎为辅，慢慢地把相关的业务往Doris里面做迁移。
　　--
　　05
　　QA环节
　　Q：把Druid迁移到Doris的成本有多大？
　　A：主要还是SQL改造的成本，可以参考POC验证阶段。
　　Q：刚才介绍的第二个场景里面的监控图可以介绍一下吗？都看了哪些指标？
　　A：Doris其实会比较关注的其实有一个就是数据导入这一块，数据导入这块最关注的是compition的效率，是否有compation的堆积，我们采用官方的默认参数。
　　Q：从指标上看Doris的实时服务在线查询性能怎么样？在数据导入情况下，性能损耗可以从这些指标上看出来吗？
　　A：实时导入这一块其实主要就是从compaction的效率来看，我们这边的一个业务场景最多的一张表6亿到10亿的数据量。另外他的峰值QPS也是能达到千到万的，导入这块压力还好。
　　Q：SQL缓存和分区缓存实际效果怎么样？
　　A：SQL 缓存这块其实还好，主要是离线场景，尤其是查询的数据量是昨天或者过去一个小时之前的这种，SQL缓存命中率会非常高。分区级缓存就是一个分区，设的是小时级别。如果这个查询涉及到的一些分区，如果在一个小时内没有数据更新，那么它就走分区缓存，如果有他可能就会走非分区级的缓存。总体来看，命中比较多的还是SQL缓存。
　　Q：Doris的查询导入合并和缓存的be节点的内存一般怎么分配？
　　A：缓存这一块我们分配得不大，还是默认的1G以内。导入这块我们其实涉及的是类似于exec_mem_limit参数，8G左右。
　　Q：可以再解释一下OLAP3.0的解决思路吗？
　　A：其实在OLAP3.0这块，业务主要的诉求其实就是大表join和导入的精准一致等。在大表join这块我们是对比了很多的引擎，Druid是偏维表，ClickHouse是偏broadcast的基于内存。
　　Q：ClickHouse和Doris应该都是近实时的，写入的数据不可能立马看到的吧？
　　A：是这样子。像我们Doris和ClickHouse的写入都是Flink直接去写，我们也没有完全做到就是说来一条写一条，还是做了一个微批次的，内存缓存150M后写入。
　　Q：方便透露了一下货拉拉目前Doris的集群情况，比如机器的数量和数据量吗？
　　A：我们集群数量还不算很多，10台左右。
　　Q：Doris的运维方面，它的便捷性和Druid、ClickHouse、Kylin还有Presto这些相比有很好的扩展性吗？
　　A：我们觉得是有的。Druid这一块我们碰到一个比较大的痛点是说它角色特别多，有六种角色，需要部署的机器也会非常多，还有就是它的外部依赖也非常多，它依赖于HDFS，离线导入的话还得有一个hadoop集群。第二个是ClickHouse对于ZK这一块也是一个比较大的依赖，且是偏伪分布式的，有点类似于数据库的分表。相对而言Doris自身只有一个FE和BE，外部依赖会很少。所以从部署的角度看，Doris会更好一些。然后再就是它的横向扩展，扩缩容也能够做到自平衡，所以相比而言会更好一些。
　　Q：在分钟级的数据且对服务性能要求比较高的实时更新场景下，可以用Doris吗？能够达到TP99 200ms吗？
　　A：和你的查询SQL有关，比如说涉及到大表join，分区数据会在10亿左右的一个数据量级，我们这一块对于查询的性能要求是在5秒以内就能够满足。如果是实时特征这一块的话，200毫秒以内可能还得再去实测一轮。
　　今天的分享就到这里，谢谢大家。
　　分享嘉宾：杨秋吉 & 张斌 货拉拉
　　编辑整理：袁洪军 孩子王
　　出品平台：DataFunTalk
　　01/ 分享嘉宾
　　杨秋吉｜ 货拉拉  大数据引擎负责人
　　张斌｜ 货拉拉  大数据工程师
　　02/ 报名看直播 免费领PPT
　　03/ 关于我们
　　DataFun： 专注于大数据、人工智能技术应用的分享与交流。发起于2017年，在北京、上海、深圳、杭州等城市举办超过100+线下和100+线上沙龙、论坛及峰会，已邀请超过2000位专家和学者参与分享。其公众号 DataFunTalk 累计生产原创文章700+，百万+阅读，14万+精准粉丝。