作业帮基于 Apache Doris(Incubating) 的数仓实践

原创
2020/04/23 21:54
阅读数 5.7W

【本文经作者授权转载,原则作者 糜利敏,联系方式见文章末尾】

关于 Apache Doris(Incubating)

Apache Doris(Incubating) 一款基于大规模并行处理技术的交互式SQL分析数据库,由百度于2018年贡献给 Apache 基金会,目前在 Apache 基金会孵化器中。

Github: https://github.com/apache/incubator-doris,欢迎大家 Star、提 Issue、Pull Request。

官方网站:http://doris.incubator.apache.org/  可以查看更多安装、部署、使用文档,也欢迎对文档内容进行校对或建议。

开发者邮件列表:dev@doris.apache.org,(如何订阅请戳这里

背景

作业帮大数据团队主要负责建设公司级数仓,向各个产品线提供面向业务的数据信息,如到课时长、答题情况等,服务于拉新、教学、BI等多个重要业务线。在过去数月内,我们通过对Doris的应用实践,构建了数仓实时查询系统。本文总结并分享下期间的工作内容,也欢迎大家一起讨论。

典型的数仓从逻辑上划分为:

大数据团队主要负责到ODS-DWS的建设,从DWSADS一般是数仓系统和业务线系统的边界。

在过去,由于缺失统一的查询系统,探索了很多模式来支持各个业务线发展。

  • 非流量
    • Kafka 业务线kafka接数据自己做数据的聚合算。主要问题在于完全没有数的概念,业务线在做大量重复的建
    • Spark + ES。每来一个业务需求,就构建一个Spark+ES集群(spark负责计算写入到ESES业务层直接使用)。效率低、构建成本高,且ES高效的使用本身本身就需要学ES的接口以及内部原理,业务线这样的精力去做
    • ES + 自定API。大数据将数据写入ES后,并case by case构建api。初步有了数的接口,但是接口不具Sql的能力,只能基于需求case by case的构建,效率太低。
    • ……
  • 流量(如pvuv等)
    • 由于数据量大,往往需要聚合,引入druid

 

这些烟囱式的系统构建方式,导致系统越来越难以维护,且业务接入效率也逐步降低。

因此,统一整个查询引擎,对于数仓建设在提高业务支持效率、降低维护成本上都具有非常重大的意义。

总体方案

经过过去数月的探索与实践,我们确立了以Doris为基础的数仓实时查询系统。同时也对整个数仓的数据计算系统做了一次大的重构,最终整体的架构图如下:

如图所示(从下到上),原始业务层日志经数据摄入系统进入数仓,在数据清洗计算层,我们将原来的Spark系统升级到了Flink,并且基于Flink-Sql提供了统一数据开发框架,从原有的代码开发升级到Sql开发来极大的提升数据的研发效率。

其后查询系统将Kafka的数据实时同步到查询引擎内,并通过OpenAPI的统一接口对外提供查询服务。

接下来,重点讲下查询系统的工作。

查询引擎选型

实时查询系统的核心在于确定查询引擎。

社区的查询引擎较多,如ImpalaPrestoDorisESxpack),以及云上的ADB等。这块考虑到调研成本、团队技术生态、维护成本等多种因素,我们最后选择了Doris 作为我们的查询引擎。

在性能调研时,我们也走了一些弯路:第一次使用Doris来做查询引擎,发现使用我们的业务Sql,延迟数据比较大,且CPU使用率很高(IDLE < 10%),

原因在于使用了AGGREGATE模型,如对于订单数据,一般会将用户支付金额等作为指标列(如一个用户从订单预订到支付,状态的改变会修改支付金额值),但是业务端的Sql中有大量的基于支付金额(指标列)的筛选查询,如统计支付金额 > 某个值的用户数。

Doris对于指标列的筛选成本较高,底层采用了类LSM-Tree的结构,因此为了确定某一行的数据是否该被筛选,需要扫描所有底层文件内包含该行的数据,进而聚合计算后才可以决策是否结果集包含该行(UNIQ模型类似)。而DUPLICATE表无法更新列。

最终使用Doris on ES,主要考虑点

  • 任意列索。基于ES的倒排索引,我可以任意列索(筛选)。个模型大大降低了业务同学的学理解成本,可以和mysql方便的构建数据模型。
  • ES的易用性以及整个技在公司内相成熟的多,维护成本低。如数据修改可以直接覆盖最新,非常简单
  • Doris on ES在数据Scan上做了大量的化操作,如列存、local优先、响应内容过滤、顺序扫描、提前终止等,对于数据的扫描性能可以达到~30w/s
  • Doris 提供了更大的Sql语法(如join、多列group by……),且整个查询过程保障了数据的准确度。大大提高了数据使用的效率和数据查询质量。
    • 由于ES缺少分布式ES-Sql需要配置size,否致返回的数据会少于期的数据

当然,对于流量分析的场景,由于指标列一般是pvuv等,业务上并没有对指标的筛选过滤需求,且Doris自身支持RollUP,因此非常适合流量类的查询分析。

因此,通过Doris我们统一了整个查询引擎端的实现,这样对于后续整个数仓的进一步建设就打下了非常重要的基础。

应用实践

基于业务场景,我们对需求进行了分类:面向业务工作台的非流量类需求以及流量分析类需求。

非流量

在实际的应用中,业务侧的需求主要分两类:

  • 细查询。教研工作台需要关注每个老的明信息,如某程的学生的到情况、预习情况……
  • 聚合查询。部门组织上会关注整个部、小内的统计信息,如到率、拉新率等

这些需求在前端查询,均需要保障低延迟。

而明细查询对于数据的时效性要求更高,因此对于明细类查询,业务侧会直接访问Doris on ES中的数据进行查询,这样基于Doris on ES的任意列检索能力可以保障业务查询模式的灵活性以及数据的时鲜性。

而对于聚合查询,由于不同指标的Sql计算的数据范围不同,且业务侧对于聚合的计算没有明细查询的时效性,因此,我们通过微批(如1min5mins10mins……)的调度能力定期计算聚合指标,并存放到ADS层的业务数据库中供前端平台查询。

为了提高数据使用效率,方便业务侧获得特定时间窗口的数据,在数据模型上,我们统一设置了Meta字段如数据更新时间,这样业务可以用来划分每次更新的数据窗口,做增量计算。

这个模式的主要好处

  • 业务端延迟可控、稳定性好。聚合查询的延迟随着具体的Sql不同而不同,定期行后的数据存放到业务层Mysql中,可以最大化可以保证查询
  • 数据修复成本低、维护方便。一旦数据有异常,可以自发对应的数据窗口行重新
    • 原来基于流式算的修数,需要从源修复,且必须驱动主流事件触,成本非常高,而基于doris on es,不同的事件可以更新不同的列或者表,只要在数据查询时join即可
  • 高性能。一般业务每次取部分列,个模式反而可以发挥ES适合大表的景以及Doris on ES列存取模式的实现,更保障了快的高性能

流量

对于流量,在数据清洗后,直接基于kafkaDoris即可,这块主要是利用Doris RollUp的能力,提供低延迟的数据查询能力

OneService

虽然上述可以初步满足业务的需求,但是从站在最终系统可持续运维的目标态来看,还有很多潜在的问题需要提供解决的空间

  • 如何保障查询稳定性
    • 多个用Sql查询,某个查询导致集群被打垮,如何快速止损
    • 多个景都在查询某一表,如何做到可控的降
  • 如何保障入的数据
    • 避免数据乱序覆盖……
    • 保障数据在多个的无、低成本迁移……如从Hive迁移到DorisES迁移到Mysql……
  • 如何提高易用性
    • 内支持Sql的系很多,如HiveHqlFlink-Sql……在部分函数法上会由于差异,如何透明的打平些差异。而不是不断的学异构
    • 数据如果跨云同步,提供多集群数据同步、查询,如何对业务透明的完成
    • 部分表需要自Rotate的能力,自动删期的数据
    • ……

上述的这些问题虽然短期内无法一一解决,但是需要提供一个能力:将来解决时控制成本,尽量做到对业务无感知。

这些都需要进一步定义出系统的接口边界,否则耦合各个系统,后续使用的用户越多,问题持续时间越久、迁移成本也越高。

因此我们设计了OneModel来统一数据模型,并且构建了OpenAPI来统一服务接口。

目前完成的功能包括

  • OpenAPI
    • Sql缓存
    • 基于业务线查询条件控制,如query_timeout
  • OneModel
    • 随着Flink的引入,合原离线的表,数据表在不同存上分布越来越多如KafkaRedisDorisHive……,因此构建<数据表,Schema,存>的元数据,支持数据表在不同存上的映射关系,一表逻辑视图,提升使用效率
    • 引入了Json-schema,保库质量符合数据模型定
  • 其他
    • Rotate Table
    • 规范化数据协议,基于数据版本解决数据写入时乱序问题
    • ……

基于上述的设计,一方面支持业务功能的同时,更重要的是切分了整个系统的接口,来降低各个系统的耦合。有几点具体的好处:

  • 数据清洗系查询基于Kafka解耦,这样查询统临时异常,不会阻塞算系。且多个Topic可以天然支持正常数据流&修数数据流的同步入
  • 业务层通过统一的接口来进行数据访问,在访问入口处可以统一方便的进行流量调度,统一的解决稳定性问题
  • 由于整个系统闭包,且接口基于数据协议耦合,定性和易用性得到了兼

应用表现

基于Doris on ES的查询系统上线数月,一直到经历了运营大促的活动,均表现出了非常好的稳定性。每天百万级次调用,99分位延迟~秒级

我们的人效也得到了了数十倍的提升:从过去一个需求进入查询系统到对外交付数据需要数人周,提升到当前模式的小时级甚至分钟级。

总结与规划

通过引入doris,解决了我们明细&聚合数据查询不统一的问题,奠定了整个数据中台在查询侧的基石,对于后续数仓向数据中台发展的路径起到了非常关键的作用。

规划

  • 跨集群实时同步。在异地多活等景下,目前缺少mysql-binlog实时同步能力,需要构建低成本的数据实时同步能力,支持在线业务定性。
  • Doris on ES多表Join性能。在尾的需求下,Join需要描两表的全部数据行内存算,尤其是大表Join大表,延就会升高。
  • Doris on ES表分区能力。如ESRotate表,目前Doris无法识别新表或者自动删除老的表映射,需要Doris表来对应ES.Index
  • Doris on ES表自同步能力。如ESSchema修改后,可以自同步到Doris
  • Doris平台化运,如建表、修改表、数据…… 

更多Doris on ES2020规划,请参见:https://github.com/apache/incubator-doris/issues/3306

在此非常感谢百度Doris团队特别是@wuyunfeng@imay 等同学热情、给力、靠谱的技术支持!!!我们也希望后续一起参与到Doris的开发建设中来!

欢迎来撩!

在线教育属于当前还在持续高速增长的业务赛道,作业帮作为一家专注于K12的在线教育公司,当前已经累计激活用户8亿+,月活1.7亿+

作业帮大数据团队致力于面向公司构建数据中台,这里可以接触到大数据下的分布式计算、存储等多种前沿的工程架构技术,欢迎各位感兴趣的小伙伴来撩~

联系邮箱:milimin@zuoyebang.com

展开阅读全文
f2c
打赏
7
64 收藏
分享
加载中
问哈 数据清洗 做了什么工作?
2020/06/11 10:18
回复
举报
将不统一的类型统一、将含义一致命名不一致的统一、计算出更贴近业务易用的字段……
2020/06/12 10:09
回复
举报
有做数据关联吗?
2020/07/23 10:45
回复
举报
针对维度表数据百万以上时,如何算pv\uv的?单纯上卷比较慢
2020/05/12 14:43
回复
举报
表行数百万以上一般没问题的。pv、uv一般都是用AGGE模型即可。我们计算2亿的uv,秒级返回,用了的rollup。所以这块得看下整个表结构如何设计的
2020/05/12 19:52
回复
举报
针对星形模型,一个大的事实表,一个百万的账户维度表,再有一些小维度表。按账号名看pv、uv,比较慢,你微信多少,聊一聊?
2020/05/13 12:07
回复
举报
你微信多少?我加你吧,拉你入doris群
2020/05/14 10:46
回复
举报
私信你了
2020/05/14 15:36
回复
举报
有群么,求拉群,跟着各位大神一起学习下
2020/06/10 16:55
回复
举报
聚合指标是通过sparksql定时做增量计算?
2020/04/30 00:22
回复
举报
不是所有的业务线同学都对sparksql非常熟悉。因此这块的架构会做的比较low level。 比如对业务线可以是一个stdin/stdout接口 至于上面跑的是一个binary、script……都可以。 底层就是一个调度器了。
2020/05/10 12:10
回复
举报
doris支持多数据源操作么 比如es 和 mysql 表的join
2020/04/27 09:38
回复
举报
支持,ES和mysql中的表join,mysql也作为doris的外表
2020/04/27 11:06
回复
举报
目前还不太了解,Doris和其他的MPP相比,有什么优势。
2020/04/26 08:33
回复
举报
morningman博主
部署简单、功能丰富、易用性强~欢迎试用~
2020/04/26 12:55
回复
举报
Good. 欢迎大家体验Doris On ES
2020/04/25 21:12
回复
举报
更多评论
打赏
16 评论
64 收藏
7
分享
返回顶部
顶部