字节跳动基于Hudi的数据湖集成实践

本文是字节跳动数据平台开发套件团队在Flink Forward Asia 2021: Flink Forward 峰会上的演讲，着重分享了字节跳动数据湖技术上的选型思考和探索实践。

文 | Gary Li 字节跳动数据平台开发套件团队高级研发工程师，数据湖开源项目Apache Hudi PMC Member

在选择了基于Hudi的数据湖框架后，我们基于字节跳动内部的场景，打造定制化落地方案。我们的目标是通过Hudi来支持所有带Update的数据链路：

需要高效率且低成本的Upsert
支持高吞吐
端到端的数据可见性控制在5-10分钟以内

目标明确后，我们开始了对Hudi Flink Writer进行了测试。这个图是Hudi on Flink Writer的架构：一条新的数据进来之后，首先会经过一个索引层，从而找到它需要去的地方。

State索引中保存了所有主键和文件ID的一一映射关系，对于Update数据，会找到其所存在的文件ID，对于Insert数据，索引层会给他指定一个新的文件ID，或者是历史文件中的小文件，让其填充到小文件中，从而避免小文件问题。
经过索引层之后，每条数据都会带有一个文件ID，Flink会根据文件ID进行一次shuffle，将相同文件ID的数据导入到同一个子任务中，同时可以避免多个任务写入同一个文件的问题。
写入子任务中有一个内存缓冲区，用于储存当前批次的所有数据，当Checkpoint触发时，子任务缓冲区的数据会被传入Hudi Client中，Client会去执行一些微批模式的计算操作，比如Insert/Upsert/Insert overwrite等，每种操作的计算逻辑不同，比如说Insert操作，会生成一个新的文件，Upsert操作可能会和历史文件做一次合并，
待计算完成后，将处理好的数据写入到HDFS中，并同时收集元数据。
Compaction任务为流任务的一部分，会定时的去轮训Hudi的时间线，查看是否有Compaction计划存在，如果有Compaction计划，会通过额外的Compaction算子来执行。

在测试过程中，我们遇到了以下几个问题：

在数据量比较大的场景下，所有的主键和文件ID的映射关系都会存在State中，State的体积膨胀的非常快，带来了额外的储存开销，并且有时会造成Checkpoint超时的问题。
第二个问题是，由于Checkpoint期间，Hudi Client操作比较重，比如说和底层的base文件进行合并，这种操作涉及到了历史文件的读取，去重，以及写入新的文件，如果遇到HDFS的抖动，很容易出现Checkpoint超时的问题
第三个问题是，Compaction任务作为流式任务的一部分，任务启动后资源就不可调节，如果需要调节，只能重启整个任务，开销比较大，如果不能灵活调节Compaction任务，就可能会出现Compaction算子空跑导致资源浪费，或者资源不足导致任务失败的情况

为了解决这些问题，我们开始针对我们的场景进行了定制化的优化

技术方案

索引层

索引的目的就是找到当前这条数据所在的文件地点，存在State中的话每条数据都涉及到一次State的读和写，在数据量大的场景下，所带来的计算和储存开销都是比较大的

字节跳动内部开发了一种基于哈希的索引，可以通过直接对主键的哈希操作来找到文件所在的位置，这种方式在非分区表下可以做到全局索引，绕过了对State的依赖，改造过后，索引层变成了一层简单的哈希操作。

写入层

早期的Hudi写入和Spark强绑定，在2020年底，Hudi社区对底层的Hudi Client进行了拆分，并且支持了Flink引擎，这种改造方式是将Spark RDD的操作变成了一个List的操作，所以底层还是一个批式操作，对于Flink来说，每一次Checkpoint期间所需要做的计算逻辑是类似于Spark RDD的，相当于是执行了一次批式的操作，计算包袱是比较大的。

写入层的具体流程是：一条数据经过索引层后，来到了写入层，数据首先会在Flink的内存缓冲区积攒，同时通过内存监控来避免内存超出限制导致任务失败，到了Checkpoint的时候，数据会被导入到Hudi Client，然后Hudi Client会通过Insert，Append，Merge等操作计算最终的写入数据，计算完成后将新的文件写入到HDFS并同时回收元数据。

我们的核心目标在于如何让这种微批的写入模式更加的流式化，从而降低Checkpoint期间的计算负担。

在表结构上，我们选择了与流式写入更加匹配的Merge on Read格式，写入的算子只负责对于log文件的追加写入，不做任何别的额外的操作，例如和base文件进行合并。
在内存上，我们将第一层Flink的内存缓冲区去掉，直接把内存缓冲区建立在了hudi client中，在数据写入的同时进行内存监控避免内存超出限制的情况，
我们将写入hdfs的操作和Checkpoint进行了解耦，任务运行过程中，每一小批数据就会写入HDFS一次，由于HDFS支持追加写操作，这种形式也不会带来小文件的问题，从而将Checkpoint尽可能的轻量化，避免HDFS抖动和计算量过大带来的Checkpoint超时的问题。

Compaction层

Compaction任务本质上是一个批任务，所以需要和流式写入进行拆分，目前Hudi on Flink支持了异步执行Compaction的操作，我们的线上任务全部使用了这种模式。

在这种模式下，流式任务可以专注于写入，提升吞吐能力和提高写入的稳定性，批式的Compaction任务可以流式任务解耦，弹性伸缩高效的利用计算资源，专注于资源利用率和节约成本。

在这一系列的优化过后，我们在一个2百万rps的Kafka数据源上进行了测试，使用了200个并发导入到Hudi。和之前相比，Checkpoint耗时从3-5分钟降低到了1分钟以内，HDFS抖动带来的任务失败率也大幅度下降由于Checkpoint耗时降低，实际用于数据处理的时间变得更多了，数据吞吐量翻了一倍，同时State的存储开销也降到了最低。

这是最终的CDC数据导入流程图

首先，不同的数据库会将Binlog发送到消息队列中，Flink任务会将所有数据转换成HoodieRecord格式，然后通过哈希索引找到对应的文件ID，通过一层对文件ID的shuffle后，数据到达了写入层，写入算子以追加写的形式将数据频繁的写入到HDFS中，Checkpoint触发后，Flink会将所有的元数据收集到一起，并写入到hudi的元数据系统中，这里就标志了一个Commit提交完成，一个新的Commit会随之开始。

用户可以通过Flink Spark Presto等查询引擎，近实时的查询已经提交完成的数据

数据湖平台侧托管的Compaction服务会定时提交Flink Batch模式的Compaction任务，对Hudi表进行压缩操作，这个过程对用户无感知并且不影响写入任务。

我们这一整套解决方案也会贡献给社区，感兴趣的同学可以关注Hudi社区最新的进展

流式数据湖集成框架的典型落地场景

流式数据湖集成框架改造完成后，我们找到了一些典型的落地场景：

应用最普遍的就是将线上数据库导入到离线数仓进行分析的场景，和之前的Spark离线链路相比：端到端的数据延迟从一个小时以上降低到了5-10分钟，用户可以进行近实时的数据分析操作。

在资源利用率方面，我们模拟了一个Mysql导入离线数仓进行分析的场景，将Flink流式导入Hudi和Spark离线合并的方案进行了对比，在用户小时级查询的场景下，端到端的计算资源大约节约了70%左右。

在字节跳动EB级数据量的数仓场景下，这种资源利用率的提升所带来的收益是非常巨大的。

对于基于消息队列和Flink构建实时数仓的用户来说，他们可以把不同数仓层级的实时数据导入到Hudi，这类数据update的情况很多，所以相较于Hive，Hudi可以提供高效率且低成本的Upsert操作，从而用户可以对于全量数据进行近实时查询，避免了一次去重的操作。

这是一个Flink双流Join的场景，很多Flink的用户会使用双流Join来进行实时的字段拼接，在使用这个功能的时候，用户通常会开一个时间窗口，然后将这个时间窗口中来自不同数据源的数据拼接起来，这个字段拼接功能也可以在Hudi的层面实现。

我们正在探索一个功能，在Flink中只将不同Topic的数据Union在一起，然后通过Hudi的索引机制，将相同主键的数据都写入到同一个文件当中，然后通过Compaction的操作，将数据进行拼接。

这种方式的优点在于，我们可以通过Hudi的索引机制来进行全局字段拼接，不会受到一个窗口的限制。

整个拼接逻辑通过HoodiePayload实现，用户可以简单的继承HoodiePayload，然后来开发自己的自定义的拼接逻辑，拼接的时机可以是Compaction任务，也可以是Merge on Read近实时查询，用户可以根据需求场景，灵活的使用计算资源。但是相比Flink双流Join，这种模式会有一个缺点，就是实时性和易用性上要差一些。