快手流批一体数据湖构建实践

本次将介绍快手为什么建设数据湖，在数据湖建设过程中遇到的问题和取得的成果，并对未来发展进行展望。

(资料图片)

一、数据湖架构：从离线数仓到湖仓一体的转变

数据建设的核心目标一般为：

① 标准统一。

② 可共享。

③ 简单易用。

④ 高性能。

⑤ 成熟安全可靠。

但是，现在常用来作为实现方案的 Lambda 架构，架构一般如下：

这里存在三个比较严重的问题：

① 离线链路时效性差。若是直接在这个链路上进行提效，则需要的成本比较高。

② 处理逻辑异构。由于目前将实时数据和离线数据分成了两个链路来处理数据，导致很多的处理逻辑无法复用。同时，也会存在一致性的问题。

③ 数据孤岛。本身多个链路的生产会存在数据孤岛，数据无法复用，并且管理相当复杂。

为了解决上述问题，快手使用了数据湖作为数据建设的一个集中式仓储方案。同时，数据湖也能够满足数据建设的核心目标。数据湖具有以下特性：

① 海量存储。

② 支持可扩展的数据类型。

③ Schema 演进。

④ 支持可扩展的数据源。

⑤ 强大的数据管理能力。

⑥ 高效数据处理。

⑦ 高性能的分析。

业内有很多的数据湖开源实现方案，快手对这些方案的基础优势及特性、社区建设情况和技术开发的可扩展程度进行了比较，最终选择了 Hudi 作为数据湖实现方案。Hudi 在摄入、处理、存储到查询，基础能力支持地比较完善，其具有的多个特点能够支持数据湖的快速构建和应用，这些特点包括：更新能力强，支持流批读写，可插拔的 Payload，支持 MOR 表类型，适配多种查询引擎，丰富的数据管理操作等。

Hudi 可以帮助快手构建更优的数据链路，去完成数据建设的核心目标，架构参考如下：

快手基于 Hudi 构建的数据湖架构具有以下优势：

① 数据 CURD。优化生产场景模型，提升了整体更新场景的时效；

② 流批读写。实现统一的处理，减少了多链路多引擎的建设成本；

③ 海量数据管理。对所有的入湖数据进行统一管理，数据平台的服务和管理方面能够复用，降低数据的使用成本。

二、基于 Hudi 快速构建快手数据湖：

建设快手数据湖遇到的挑战以及解决方案

如何使用 Hudi 建设达到核心目标，需要先了解 Hudi 的基本能力：

① 支持不同类型的写入方式：特别是通过增量写入和数据合并的两个基本操作，实现快照的生成；

② 可插拔：可支持所需要的更新逻辑，比如定制化更新模式，可以基于此进行扩展应用场景；

③ 表类型：正如前面提到的，增量写入和数据合并的操作共同组成快照更新。这两种基本操作的实现决定了表的类型。不同类型的表，作用不同的应用场景，比如写多读少的情况下，选择使用 MOR 更实时和节约资源；

④ 元数据统计：因为 Hudi 本身实现了更新能力，甚至在之上实现一部分的业务逻辑的，需要保障可描述、可追溯的能力。所以通过元数据的收集和应用，来保证数据的可追溯性；

⑤ 读取方式：支持 Hadoop 的 inputformat 的方式，兼容常用的查询引擎，比如spark、trino 等。

使用这些能力，可以为生产链路实现提效与统一。

提效主要还是在优化构建离线数仓的时间：

① 比如分层建设时，需要先同步数据，然后再使用离线清洗，再生成后续的数仓的加工数据。现在可以直接一步通过 Flink 任务清洗实时数据，然后使用 Hudi 多级动态分区同步。

② 还有，在离线链路生产时，有些数据生产是有更新逻辑的，比如更改部分数据内容。在老的架构下，需要将所有数据都读取一遍，然后将修改了某几列的新数据再完全写入。这里不但时效很差，而且成本消耗很大。现在可以利用 Hudi 的更新能力来高效地更新列数据。

③ 其他的，比如活动的数据需要进行快照分析时，离线链路都是小时级别的延迟，一般都需要使用实时链路同时生产。使用 Hudi 就可以进行准实时快照的构建，并提供兼容的查询。

统一的实现，主要是选用了 Flink 引擎作为流批一体的计算引擎，在整体 Hudi 数据湖的生产中进行应用。

通过 Hudi 数据湖架构建设的数据链路如下所示：

快手在通过 Hudi 数据湖架构建设新的数据链路中，遇到了许多问题。下面，介绍一下快手在建设数据湖过程中遇到的 5 个重要问题以及具体的解决方案。

1、数据摄入的瓶颈

问题描述

快手的数据链路都是基于 Flink 生产的，其 Hudi On Flink 架构如下图所示。

采用上述架构进行数据生产时会遇到性能瓶颈。由于写入多分区的数据时会通过 BucketAssigner 来进行数据分发，再使用 BucketWriter 实现缓存写入，那么，当 BucketWriter 之间数据不均衡时，写入会频繁触发溢写。而当溢写发生时，又会产生背压。另外，在提交数据时，由于 BucketWriter 与 Flink 快照进行了绑定，所以 Flink 快照无法实现整点触发。

解决方案

为了解决上述提到的写入瓶颈问题，快手优化了写入逻辑，主要应用于增量数据的同步链路。首先，优化写入模式以提升性能。Flink 写入方式从缓存写入修改为流式写入。写入数据时不需要缓存，而是直接落盘，并且支持单生产者多消费者的模式，每一个分区文件都可以并行写入。这样，可以提高 CPU 的使用率。其次，在摄入的过程中对分发逻辑进行了优化，实现了一个动态感知的模块。该模块用于统计数据流量，均衡分发数据到写入节点，从而保证了各分区之间的数据均衡，来避免某个写入节点受到过大的数据压力。

为了实现数据的整点提交，快手实现了自动分区发布功能。根据数据的时间戳生成了分区时间，并且在摄入过程中实时上传数据的轮转的时间。在中心协调器里面实现了一个判断逻辑，如果所有的节点均已完成轮转，则进行快照的触发，完成最终的数据提交和分区的同步，来实现整点级的分区发布。

在上述架构中，算子可以进行横向扩展，整体吞吐量比社区版本提升 10 倍以上，并且能将文件控制在需要的大小（例如：256M）左右。

2、无法使用数据时间进行快照查询

问题描述

在准实时的数据链路上，需要使用 Hudi 的 Time Travel 功能来实现快照查询。但是，在 SQL 查询是使用 Timeline 的时间点来进行定位的，而 Timeline 的时间与数据时间不同，且具体的 Timeline 的提交时间在存储时无法准确感知。

解决方案

快手在 Hudi 的 Time Travel 功能上增加了一个时间版本的元信息。每次写入时，会通过数据的时间字段来计算数据的版本号。与分区发布过程相同，会实时上传版本的轮转时间。在中心协调器判断是否所有分区已经完成了轮转，以快照触发。

由于在提交时存储快照的数据版本信息，在查询时，SQL 可以直接使用版本信息来进行查询。在构建输入快照的过程中间，会读取 TimeTravel 的提交信息。这样，通过判断数据版本信息是否小于等于 SQL 中指定的时间戳的版本号来构建增量快照，实现某一个时间点的快照查询。

3、Flink On Hudi 的更新瓶颈

问题描述

在使用 Flink 引擎生产 Hudi 表的过程中，更新是存在一定的瓶颈的。主要体现在，对 Hudi 的不同操作使用的资源是错配的。比如，写入操作的写入内存一般就是摄入的缓存大小。而对于合并操作，合并过程会根据增量数据的数据量来决定 compaction 所需要的内存，一般情况下，这个内存占用量是大于缓存空间的。清理操作中，在构建 FileSystemView 对象时，所占用的内存比较大。

然后，混合操作会影响增量写入的稳定性。比如合并过程中，并发度无法进行扩展，会导致运行时间长，进而导致快照产生时间延迟（因为快照触发是需要水位（watermark）下推），甚至会导致任务超时。清理的时候如果遇到异常，也会导致任务的失败。因此，操作之间的资源复用对操作的执行进度会有影响。

解决方案

解决问题的主要工作是将操作进行分离，支持多种操作并行执行来构建 Hudi 的数据源。

首先，Hudi 支持多种索引。在快手活动期间，会选用 State Index，配置 TTL 来保存一定时间内的快照结果。在需要并发写入的任务中，由于任务的索引需要相互感知，因此会选用 Bucket Index，可以有效控制写入缓存资源的占用，而且可以在外部进行操作的运行管理。在表创建时，触发生成和合并的调度作业；表下线时，自动下线挂载的调度作业。

此外，多个数据源的写入还需要实现并发控制。首先，对元数据进行加锁，来避免对元数据的并发操作。然后，在支持并发写入的过程中，支持了关联引用，为合并的功能增加了占位逻辑，后续的写入基于占位合并的 instant，在合并完成之后，基于合并的写入也是对外可见的，这种方式可以提高写入的吞吐量。此外，也支持开启 OCC 控制的并发写入，在写入相同的 base 文件时进行并发检查，支持冲突的失败回滚或者合并处理，以防止数据结果不正确的现象出现。

4、多任务合并能力不足

问题描述

多任务合并宽表时，在多任务并发运行写入的场景中，进行索引选择时，需要考虑到索引数据需要被多任务感知到的因素。若是采用外部索引，则使用成本较高。若是采用 Bucket Index 作为索引，则进行多任务并发写入时，性能上有优势。但是，存在一个合并时的瓶颈，这是由于一般情况下，Bucket Index 使用文件大小来控制计算桶数，而合并时使用的资源又取决于增量文件的数据大小，这会导致合并任务的并发度较小，无法满足合并时的性能需求。在合并的过程中，存在读取和更新的操作，若是过程中出现了溢写现象，则整个合并速度会很慢。

在 Schema 的使用方面。不同的写入任务会使用不同的 Schema，而合并时依赖于写入任务的 Schema 来生成合并的 Schema 以生成最终的 Base 文件。但是，一些自动上线的写入任务无法被合并作业感知到。

解决方案

快手实现的并发写入作业支持了逻辑分桶和多类型合并的能力。逻辑分桶是在物理桶的组织之上进行了二次哈希，本质上是将物理桶分成了更多的桶，在需要写入时，要先进行桶的排序，并创建对应的索引文件。在后续的合并过程中，基于逻辑桶来生成合并计划，每一个逻辑桶都会生成一个对应的算子实例。

合并时，作业先读取物理桶的数据，然后通过索引 seek 到对应逻辑桶的数据位置，之后进行可选择类型的合并。一般地，在写入并发已知的情况下，sortMerge 是更快的。在元数据中，增加了合并 Schema 的配置，在写入时将 Schema 更新到数据源，从而实现了合并Schema的自动扩展和合并任务的自动感知生产。

5、Hudi 生产保障困难

问题描述

Hudi 作为一个较复杂的架构，从生产到运维有比较丰富的支持，比如不同模块有对应的配置类，支持 metrics 系统、支持 Hudi-Cli 查询元信息。

但是，这些功能支持在实际生产环境的使用效果并不好。首先，Hudi 的配置过多，使用起来很麻烦。其次，在监控报警和异常中断的能力上，作为线上服务略显不足，需要进行自定义强化。因此，快手针对线上需要，加强了生产的保障能力。

解决方案

① 配置精简。只需要设置一些基本参数（SQL 方式），比如任务类型、保存时间、提交间隔，就可以自动推导生成其他的配置参数。

② 一致性保障。在一致性保障方面，快手自定义实现了 PreCommit 检验模块，比如，会对增量数据的输入输出条数进行校验，同时数据块的写入情况在提交之前也会做校验。

③ 稳定保障。在稳定性方面，快手完善了不同算子的 metrics，包括耗时和数据处理情况，流量吞吐情况等。同时还监控着分区分布，耗时，任务的指标监控来共同保障生产的稳定性。

三、快手的实践案例

快手数据湖在构建完成之后，在一些具体的业务上进行了应用，取得了明显的收益。

下面使用四个比较典型的案例，来对比基于数据湖建设的新数据链路与旧数据链路之间的差异。

最早的时候，核心数仓的 DWD 层生成是需要多层的离线调度任务来进行。在切换到 Hudi 之后，就可以准实时的生成 DWD 层的动态更新数据。此外，还可以根据需要来选择性的进行数据重分布的操作来对接下来的读取操作进行提效。这个场景上，快手将离线链路升级成了准实时的链路，在计算资源上持平，时效上有 50% 以上的提升。

数据在应用过程中，还会有活动数据快照查询的需求。早期，这些数据若要使用多个数据源进行生产和查询，需要用到离线链路，这种方式的时效性很差，一般会达到小时级。如果想使用实时链路加速，需要比较复杂的处理过程。切换到 Hudi 之后，将离线快照的更新时效从小时级缩短到了分钟级，整体时效达到十多分钟左右，而且计算资源比以前节省了 15%。在后续的查询过程中，可以对离线桶的快照数据进行关联查询，最终生成需要的活动的结果数据。

生产过程中的数据留存场景中，在生产留存数据时，最早的生产流程是需要用多天的日活数据去重复地生产标签表，然后与日活的数据 JOIN 生产到最终日活表内，这个过程涉及多次的日活表的读取和全量数据的回收。切换至 Hudi 后，通过将日活留存状态直接更新至留存表，数据生产模式从多次的合并生产转换成了单表生产。在使用当日的日活数据去更新留存表，之前的数据是已经存在的，只需要将日活数据去更新留存状态即可。这个场景下，链路的生产方式上的优化，整体计算资源由于全量读写到增量写入的转换，根据需求进行定时合并，时效上也有 50% 的提升。

在特征的生产场景内，上游的多个数据生产点合并生产出宽表结果。这个场景下，原来是使用 HBase 来进行合并，在 HBase 中进行行存，在外部用 Hbase 进行生产有一个额外的维护成本，并且需要使用到 HBase 的导入导出工具来进行离线操作，生产链路较长。切换至 Hudi 后，可以直接复用已有的生产链路逻辑，然后直接对摄入到 Hudi 表内的数据基于并发合并能力构建一张宽表。这个过程中，也可以保证数据是有序的。例如，读取时可以根据数据需求，比如上游增量写入数据源的数据已经就绪了，下游就可以直接进行导入。可以理解为实时感知，可以提升整体的处理时效。

可以的观察结果为，目前的离线分批的合并升级成了准实时单表合并，对时效升级明显。首先，处理链路的计算时间缩短，最长时间节省 5 个小时；在链路计算过程中所占用的临时存储空间和计算资源得到了节省，同时，也节省了 HBase 集群所需要的开销。

四、快手的发展规划

快手数据湖当前还有一些待优化的工作。首先，缺少完善的元数据和数据管理服务。在查询模式上，由于不支持实时表，也还没有达到离线和实时查询的统一。此外，当前快手数据湖生产方式还没有做到无感知的兼容，所以主要在新的场景上使用，总体的使用率占比不高。

未来快手数据湖将作为统一存储的技术组件，支持更多类型的数据以及拓宽数据湖支持的表类型，例如实现类似于实时表的定义。完善数据的管理来提升数据组织的合理性。实现兼容已有链路的轻量切换方案。将实现流批一体的数据生产，提供高效统一的查询能力作为数据湖建设的最终愿景。

五、问答环节

Q1：详细介绍一下加锁的部分。

A1：其实社区本身也支持 OCC 机制，其实现逻辑是，在写入过程中对元数据进行加锁，在最终的提交阶段，会对写入文件做 CAS 操作，通过对比来发现冲突。若是发现两个写入任务写入同一个 base 文件的情况，即表示写入任务之间存在冲突，会将后写入的作业标记为失败。快手也使用了这个机制来避免并发写入时，写入同一个 base 文件，影响最终结果。

快手在这个基础之上，对合并和更新过程做了一个优化。比如说，Flink On Hudi 架构的准实时写入过程中，若按照社区的写入逻辑，将合并和更新识别为两种操作，会导致合并阻塞了整个写入操作，或者合并操作一直无法完成，会导致读取数据源的效率低。经过快手的优化后，写入过程中，写入效率不会受到 Compaction 结果的影响。因为，快手会使用之前合并执行计划的基于时间戳的占位符来进行写入操作。社区的默认逻辑是基于 baseTime 来生成数据，这使得合并的结果和写入数据之间可能存在冲突。若是采用占位的合并计划的 Instance 与已有的数据不会产生冲突。但是需要保证合并操作必须完成，否则后面的写入数据是不可见的，通过这种方式，可以提升整体的增量写入的吞吐量。

Q2：Compaction 资源错配的问题怎么解决？异步的 Compaction 是否有相关的经验可以分享一下？

A2：资源错配的主要原因是写入操作需要的资源和并发资源不一致。若是将写入过程各个操作分离开，那么可以根据写入任务的流量情况来调整写入资源。快手目前仍然采用默认的配置，写入操作的一个 TaskManager 占用了 6G-8G 的内存，其中包含了多个并发写入的 Slot。在合并过程中，取决于需要的时效。比如说，需要整体的合并时效比较高的话，需要尽量的避免溢写现象的发生，这时，需要将 Compact 的内存设置得比较大。默认情况下，快手将合并任务调整为 4 核 10G 左右。当有写入数据量比较大且合并速度快的需求时，则需要将内存设置更大一点，这样增量数据基本上存储在合并操作的内存中。这样，1 到 2G 的文件可以在 10 分钟以内处理完成。

标签：