01

概述

增量处理是一种提取、加载和转换 （ELT） 数据子集以使最终结果保持最新状态的技术，已成为为数据湖仓一体构建数据管道的标准。与传统方法不同，传统方法通常涉及拉取完整的数据快照以进行存储覆盖或使用昂贵的联接操作来识别更新，而现代数据湖仓一体通常使用本质上支持增量处理的存储格式来简化架构。得益于原生支持，Medalion架构已广受欢迎，并已被许多公司用于生产管道。这种架构的特点是三个关键层：青铜层，对后处理需求至关重要；白银层，保证数据质量；以及黄金层，提供商业价值。

Medalion 架构：从应用程序到 AI 和 BI

在接下来的部分中，我们将讨论 Hudi 如何实现增量处理，这非常适合支持 Medalion 架构的稳健实现。

02

增量查询

Hudi 通过在 Timeline 中持久保存提交元数据来有效地跟踪事务日志形式的更改，从而自然而然地促进了增量处理，在大多数情况下，增量处理依赖于基于时间戳的检查点。Hudi 的增量查询功能是通过以下配置启用的：

hoodie.datasource.query.type=incremental

hoodie.datasource.read.begin.instanttime=202305150000

hoodie.datasource.read.end.instanttime=202305160000 # 可选参数

通过上述参数，便可以检索在指定时间窗口内更改的数据。有关更多的使用示例，请查看文档。有几点需要在使用中注意：

• 设置 hoodie.datasource.read.begin.instanttime=0 将请求从表历史记录的开头开始，对表所做的所有更改。

• 省略了 hoodie.datasource.read.end.instanttime 将会获取表中截止目前完成的提交的所有更改。

• 增量查询返回的数据包含在指定时间范围内更新的记录[1]。这些记录与表中最新完成的提交对应的版本匹配。如果设置了 hoodie.datasource.read.end.instanttime，则记录将与此指定结束时间表示的提交保持一致。

• 当开始时间设置为 0 且省略结束时间时，增量查询实际上等同于快照查询，检索表中所有的最新记录。

现在我们已经了解了增量查询的行为，我们将深入研究细节。下图显示了从 Hudi MoR 表中获取增量数据所涉及的工作流。

Hudi 增量查询流

增量查询遵循第二篇中描述的读取流程，实现两个内部 API：collectFileSplits() 和 composeRDD()。实现主要分为以下几个步骤：

• collectFileSplits() 负责识别与查询相关的所有文件。此函数根据用户输入推导开始和结束时间戳，以定义特定的时间范围。然后，此时间范围用于过滤时间线上的提交。

• Hudi 的时间线由一系列事务日志组成，本质上代表了随时间推移所做的更改。在指定的时间范围使用 composeRDD() 函数处理所需的相关文件。

• 在 Hudi 表中，每条记录都包含一个名为 _hoodie_commit_time 的元数据字段，该字段将记录链接到时间线中的特定提交。在为记录加载目标文件的过程中，增量查询会构建一个提交时间筛选器，以进一步减少读取的数据量。此过滤器被推送到文件读取级别，允许 composeRDD() 优化仅加载那些打算返回的记录。

03

变更数据捕获 (CDC)

增量查询可以有效地显示哪些记录已被更改及其最终状态。但是，它们没有提供有关这些更改性质的具体详细信息。例如，如果记录 X 被标识为已被修改，则增量查询不会返回在更新之前的列值，也不会阐明它是否是新插入的记录。此外，它不会表明是否有任何记录被硬删除。为了解决这些限制，Hudi 0.13.0 引入了变更数据捕获 （CDC）。这种增强的增量处理模式提供了更全面的数据修改视图，包括插入、更新和删除，从而可以更清楚地了解数据集中的更改。

要启用 CDC 功能，用户需要设置此 table 属性 hoodie.table.cdc.enabled=true[2]。写入器将遵循此设置，并激活与 Base Files 一起创建 CDC 日志文件的过程。由于 Hudi 的文件分组机制，这些 CDC 日志文件包含在保存更改数据的同一文件组中。这使得扩展表服务（如 Clean）变得容易，也方便实现恢复性的操作（如 Restore），因为能同时管理 CDC 日志文件和数据文件，以实现更连贯的文件管理。

要拉取 CDC 数据，用户只需将增量模式设置为 CDC 即可。与时间范围相关的行为仍然适用于 CDC 模式。

hoodie.datasource.query.type=incremental

hoodie.datasource.query.incremental.format=cdc

hoodie.datasource.read.begin.instanttime=202305150000

hoodie.datasource.read.end.instanttime=202305160000 # 可选的

下图简要概述了写入器和读取器如何与 CDC 文件和数据交互。

Hudi CDC 写入和读取流程示意图

在写入器端，Hudi 的写入句柄保存信息[3]关于写入记录的预期操作（INSERT、UPDATE 或 DELETE）。然后，此信息被编码为特定的 CDC 日志文件格式，其中包含四个字段，如图所示。可为 的“before”和“after”字段存储更改前后的完整记录快照。用户可以灵活地通过调整 hoodie.table.cdc.supplemental.logging.mode 来减少记录的数据量：DATA_BEFORE 模式用于跳过“after”字段，或设置 OP_KEY_ONLY 模式为存储记录键而不是“before”和“after”字段。

在读取器端，加载 CDC 日志文件以构建结果，遵循类似于普通增量查询（其增量模式称为 latest_state）的过程。如果同时记录了“before” 和“after”字段，则结果将直接从 CDC 日志文件中提取。如果使用不太详细的日志记录模式，则将通过查找表中的现有记录来动态计算结果。这实质上是在节省存储空间和运行 CDC 查询的效率之间进行权衡。

更丰富的信息

CDC功能的引入极大地增强了 Hudi 表的使用，支持更广泛的场景并提供有价值的信息。以频繁的借记和贷记交易的账户余额为例。如果没有 CDC，定期快照查询或 latest_state 模式的增量查询只能看到余额的微小变化或没有变化，可能会错过关键波动。通过 CDC 查询，所有更改都会被揭示出来，从而提供账户活动的全面视图。这种级别的详细信息对于使欺诈检测算法能够采取相应的措施至关重要。

04

回顾

在本文中，我们简要介绍了增量处理和 medallion 架构

[1]此处的更新操作和时间窗口对应于 Hudi 表中的处理时间，而不是业务域中的事件时间。

[2]启用后，用户将无法在表的生命周期内打开和关闭该设置。强制实施此限制是因为它会影响存储布局，并且用例不希望适应灵活性。

[3]根据执行引擎和索引配置，写入器或压缩运行程序都可以访问此信息。