京东实时数据仓库开发实践
The following article is from DataFunTalk Author 王日宇
导读：本文主要介绍京东实时数据仓库技术的过去和未来，使用Delta Lake完成离线数据的增量更新，建设批流一体开发分析体系简化传统数据仓库架构，以及京东的业务场景在数据湖上的落地经验和技术挑战。
01
1. 传统数据仓库的架构
首先介绍一下我们传统数据仓库的架构，目前主流的离线数据仓库是基于分布式存储分层的Lambda架构。如上图所示，由上下两条链路构成，上面链路代表离线层的处理，下面链路代表实时层的处理。这个链路既是现在设计架构上的链路，也是业务数据流转的链路，同样也是我们日常开发维护的链路。这套体系架构奠定了我们大数据分析的基础，也取得了很多收益。但随着技术的发展和业务上对实时性的要求越来越强，尤其近几年实时计算发展的特别快，现有的这套架构逐渐暴露出一些弊端。
基于Lambda架构建设的实时数仓，第一条是针对于实时性要求高的业务系统 ( 通常是秒级 ) 的数据流转链路，另一条就是传统意义上的离线计算 ( 通常是天级 ) 的数据流转链路，甚至有些业务系统还会有准实时计算的数据链路 ( 例如小时级延迟 )。不同业务系统，根据不同的时效性去选择和设计数据处理加工方式。
2. 在传统数据仓库实践中遇到的问题
① ACID语义性无法保证
简单来说就是无法做到一边写入一边读取，我们目前更多依赖读写任务在调度时间上的错配来解决读写冲突，保证数据一致性。
② 离线入库潜在的不可靠性
离线数据加工任务一般是T+1的，今天的生产数据，需要第二天凌晨抽取到大数据机房，然后进行后面的业务计算。有些业务系统的数据可能分布在全国各地数千个MySQL数据库中，假如其中某几个数据库出现问题，那么离线数据就会造成缺失，从而影响后面的数据分析计算的准确性。
③ 细粒度的数据更新功能缺失
Hive中不支持update、delete这种细粒度操作，即使只更新Hive表中的某几条数据，都需要重写整个表，或至少重写整个分区，而一个分区就是一天的数据，整个操作就需要先读取一天的数据，然后计算后再写回去。这样的话，他所需的执行时间，读写数据量，资源消耗都是比较大的。
④ 数据流转路径复杂
很多情况下，处理离线数据和实时处理的数据逻辑都是一样的，只不过需要面向不同的场景。比如说离线要使用数据做更复杂的分析，实时需要做一些秒级或毫秒级的查询。这样的话，当业务逻辑有变化时，实时需要更新一次，离线还需要更新第二次，两条链路对应两份数据，很多时候，实时链路的处理结果和离线链路的处理结果甚至对不上。
上面就是针对目前数仓所涉及到的四个挑战的大致介绍，因此我们也是通过对数据湖的调研和实践，希望能在这四个方面对数仓建设有所帮助。接下来重点讲解下对数据湖的一些思考。
02
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的各自特点。
03
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的支持
历史版本回溯和审计
抽象存储接口
查询性能提升
04
使用Delta实时数据湖后我们的开发流程可以简化如下：
如图所示，与上面的Lambda架构相比，只有一条数据流转链路。首先将业务数据库的binlog日志实时的写入kafka，然后通过SparkStreaming实时消费kafka中的数据，解析binglog日志后落入Delta数据湖中，因为整体的落数过程是实时的，所以下游既可以实时流处理也可以离线批处理。这样可以降低开发成本和存储成本，而且如果遇到脏数据的写入，整个回滚和Debug过程也会很方便。
05
最后做一下简单的总结：
Delta本身刚开源不久，内部有很多优秀特性没有开源出来，如直接使用SQL进行版本回溯，DFP动态文件裁剪，还有Z-Ordering，使用一些策略来优化数据存储分布，提高下游数据的查询效率等。
小文件和历史文件的清理问题。Delta每次写入数据时都要写一批小文件，HDFS对小文件是非常敏感的，如果小文件过多，namenode的压力会特别大。
Hive Connector的支持。社区的Hive Connector绑定的Spark Delta版本都是紧耦合的，有一些API的接口都不一样，需要自定义改造Hive Connector，支持生产环境上的版本。
计算引擎和使用方式的支持。这一点主要是突出在Hive和Presto的使用上，无论是Hive还是Presto，如果想读一个Delta表的话，必须新建一个名字不一样的外部表，location指向Delta表的位置，这样对用户侧来说，读同样的数据，存在多个不同的表名，用起来会不太方便。
今天的分享就到这里，谢谢大家。
关注不迷路~ 各种干货、资源定期分享！
🧐分享、点赞、在看，给个3连击呗！👇