京东实时数仓开发下的开源数据湖选型

王日宇 2021-08-18 10:14:43

导读:本文主要介绍京东实时数据仓库技术的过去和未来,使用Delta Lake完成离线数据的增量更新,建设批流一体开发分析体系简化传统数据仓库架构,以及京东的业务场景在数据湖上的落地经验和技术挑战。

 

一、传统数据仓库面临的挑战

 

 
1. 传统数据仓库的架构

 

图片

 

首先介绍一下我们传统数据仓库的架构,目前主流的离线数据仓库是基于分布式存储分层的Lambda架构。如上图所示,由上下两条链路构成,上面链路代表离线层的处理,下面链路代表实时层的处理。这个链路既是现在设计架构上的链路,也是业务数据流转的链路,同样也是我们日常开发维护的链路。这套体系架构奠定了我们大数据分析的基础,也取得了很多收益。但随着技术的发展和业务上对实时性的要求越来越强,尤其近几年实时计算发展的特别快,现有的这套架构逐渐暴露出一些弊端。

 

基于Lambda架构建设的实时数仓,第一条是针对于实时性要求高的业务系统 ( 通常是秒级 ) 的数据流转链路,另一条就是传统意义上的离线计算 ( 通常是天级 ) 的数据流转链路,甚至有些业务系统还会有准实时计算的数据链路 ( 例如小时级延迟 )。不同业务系统,根据不同的时效性去选择和设计数据处理加工方式。

 

 
2. 在传统数据仓库实践中遇到的问题

 

1)ACID语义性无法保证

 

简单来说就是无法做到一边写入一边读取,我们目前更多依赖读写任务在调度时间上的错配来解决读写冲突,保证数据一致性。

 

2)离线入库潜在的不可靠性

 

离线数据加工任务一般是T+1的,今天的生产数据,需要第二天凌晨抽取到大数据机房,然后进行后面的业务计算。有些业务系统的数据可能分布在全国各地数千个MySQL数据库中,假如其中某几个数据库出现问题,那么离线数据就会造成缺失,从而影响后面的数据分析计算的准确性。

 

3)细粒度的数据更新功能缺失

 

Hive中不支持update、delete这种细粒度操作,即使只更新Hive表中的某几条数据,都需要重写整个表,或至少重写整个分区,而一个分区就是一天的数据,整个操作就需要先读取一天的数据,然后计算后再写回去。这样的话,他所需的执行时间,读写数据量,资源消耗都是比较大的。

 

4)数据流转路径复杂

 

很多情况下,处理离线数据和实时处理的数据逻辑都是一样的,只不过需要面向不同的场景。比如说离线要使用数据做更复杂的分析,实时需要做一些秒级或毫秒级的查询。这样的话,当业务逻辑有变化时,实时需要更新一次,离线还需要更新第二次,两条链路对应两份数据,很多时候,实时链路的处理结果和离线链路的处理结果甚至对不上。

 

上面就是针对目前数仓所涉及到的四个挑战的大致介绍,因此我们也是通过对数据湖的调研和实践,希望能在这四个方面对数仓建设有所帮助。接下来重点讲解下对数据湖的一些思考。

 

二、实时数据湖的探索和经验

 

 
1. 数据湖开源产品调研

 

数据湖大致是从19年慢慢火起来的,目前社区主流的开源产品主要有三种:Delta、Hudi和Iceberg。它们在功能实现上各有优劣,接下来简单对比一下。

 

图片

 

上表是一个简单的社区热度统计:

 

  • Delta Lake:在17年的时候DataBricks就做了Delta Lake的商业版,主要想解决的也是Lambda架构带来的数据存储和控制问题。

 

  • Hudi:支持快速的更新以及增量的拉取操作,包含copy on write 和 merge on read 两种表格式。

 

  • Iceberg:的初衷是想做一个标准的Table Format,代码抽象程度比较高,社区也正在进行Flink的读写支持。

 

 
2. 选择Delta lake的原因

 

下面这个表格例举了部分功能点的对比,这些都是我们在做技术选型时比较关心的几个点。比如说ACID特性,历史回溯,多版本并发控制等。

 

图片

 

当时我们团队也在技术方案选型上讨论了很久,使用不同的应用场景做了不同方面的测试,最终选择了Delta。首先是因为功能完整性上比较符合我们的要求;其次我们本身将数据湖定位成基于离线计算的数据存储更新服务,再加上我们团队本身就承担着spark的基础研发工作,比如常见的sql查询优化,shuffle优化等等,对spark的了解会比较深入一些,所以我们最终选择Delta作为数据湖的基础,同时开发过程中吸取Hudi和Iceberg的各自特点。

 

三、Delta Lake核心原理

 

 
1. Delta Lake简介

 

引用来自官网对于Delta lake的一段介绍"Delta是一个开源的带有ACID语义的存储控制层,其中Delta的数据表主要是由数据文件和事务日志两部分组成。"

 

图片

 

如图所示,可以看到这是Delta表物理上的文件结构的组成,比如说我们有一个my_table表,与常规的离线Hive表不同的是,它下面会有一个_delta_log目录,这个_delta_log我们叫做Transaction log,也就是事务日志,然后就是常规的数据文件,数据文件的格式是parquet,日志文件的格式是普通的json格式。

 

Transaction log是整个Delta核心,也是所有Delta功能实现的基础,所有对Delta的操作,无论是增删改还是修改表结构,都会被记录到Transaction log中。所以我们接下来重点介绍一下Transaction log是什么。

 

 
2. 事务日志解析

 

图片

 

Transaction log主要涉及到三方面的信息:when,who,how

 

  • 一次事务就是一次commit,日志中会记录commit的基本信息,简单来说就是是谁在什么时候怎么做的commit,以截图中的日志为例,会有一个时间戳1600071805932来记录什么时候的commit,是STREAMING UPDATE做的commit的,commit内容是新增了8个数据文件。

 

  • 把涉及到的具体文件路径和统计信息写到log中,比如说他的文件名是什么,每个文件的大小是多少,是什么时间修改的,它都会记录。

 

  • 表的Metadata信息,字段名、字段类型、文件格式、配置属性等。这些与普通Hive表存在metastore里的内容是完全一样的。

 

 
3. Delta数据表读取流程

 

以一次添加数据的操作为例,简单介绍一下log的具体内容,以及Delta数据表的读取流程。

 

图片

 

Delta每次更新都会形成一个log,一系列的更新操作也就形成了多个log。log的命名是严格按照版本号递增的顺序命名的。Delta内部为了提高读取性能,每10个log会生成一个checkpoint文件,每次checkpoint都会把最新的checkpoint文件路径记录到_last_checkpoint文件中,这样随着时间的迁移整个表的变更操作都会被记录下来。

 

Checkpoint简单来说是前面所有json log的总和,但并不是简单的堆在一起,他包括消除一些冗余信息的合并操作。比如说在3版本中新增了两个文件A和B,在10版本中删除了文件A,那么这个表就只剩下文件B了,此时checkpoint只会记录文件B,再加上本身checkpoint使用parquet列式格式保存,spark读取性能会提高很多。

 

以图中左边的例子为例,总结一下Delta数据表具体读取流程:

 

① 先使用_last_checkpoint找到最近的checkpoint文件,也就是图中的000010.checkpoint.parquet。

 

② 再找到checkpoint文件之后的json log文件,就是图中的11版本和12版本的json。

 

③ 最后合并所有json log和checkpoint log的记录,得到数据表在该版本状态下包含哪些具体的数据文件。

 

 
4. Delta Lake特点

 

有了Transaction log后,很容易实现下面一些特点:

 

  • 支持批流读写

  • 提供ACID语义

  • Update/delete的支持

  • 历史版本回溯和审计

  • 抽象存储接口

  • 查询性能提升

 

四、批流一体开发流程

 

使用Delta实时数据湖后我们的开发流程可以简化如下:

 

图片

 

如图所示,与上面的Lambda架构相比,只有一条数据流转链路。首先将业务数据库的binlog日志实时的写入kafka,然后通过SparkStreaming实时消费kafka中的数据,解析binglog日志后落入Delta数据湖中,因为整体的落数过程是实时的,所以下游既可以实时流处理也可以离线批处理。这样可以降低开发成本和存储成本,而且如果遇到脏数据的写入,整个回滚和Debug过程也会很方便。

 

五、总结

 

最后做一下简单的总结:

 

Delta本身刚开源不久,内部有很多优秀特性没有开源出来,如直接使用SQL进行版本回溯,DFP动态文件裁剪,还有Z-Ordering,使用一些策略来优化数据存储分布,提高下游数据的查询效率等。

 

小文件和历史文件的清理问题。Delta每次写入数据时都要写一批小文件,HDFS对小文件是非常敏感的,如果小文件过多,namenode的压力会特别大。

 

Hive Connector的支持。社区的Hive Connector绑定的Spark Delta版本都是紧耦合的,有一些API的接口都不一样,需要自定义改造Hive Connector,支持生产环境上的版本。

 

计算引擎和使用方式的支持。这一点主要是突出在Hive和Presto的使用上,无论是Hive还是Presto,如果想读一个Delta表的话,必须新建一个名字不一样的外部表,location指向Delta表的位置,这样对用户侧来说,读同样的数据,存在多个不同的表名,用起来会不太方便。

 

今天的分享就到这里,谢谢大家。

 
作者丨王日宇
来源丨公众号:DataFunTalk(ID:datafuntalk)
dbaplus社群欢迎广大技术人员投稿,投稿邮箱:editor@dbaplus.cn
最新评论
访客 2021年09月03日

有没有1000多张表

访客 2021年08月28日

metrics =》 metrix 错误

访客 2021年08月25日

只看到如何避免,如何减少书写慢 sql

访客 2021年08月25日

没看到如何治理呀

访客 2021年07月23日

果然k8s不是神!

活动预告