Apache Hudi 批处理模式

Apache Hudi(简称：Hudi)使得您能在hadoop兼容的存储之上存储大量数据，同时它还提供两种原语，使得除了经典的批处理之外，还可以在数据湖上进行流处理。

1. 现状说明

1.1 数据湖摄取和计算过程 - 处理更新

在我们的用例中1-10% 是对历史记录的更新。当记录更新时，我们需要从之前的 updated_date 分区中删除之前的条目，并将条目添加到最新的分区中，在没有删除和更新功能的情况下，我们必须重新读取整个历史表分区 -> 去重数据 -> 用新的去重数据覆盖整个表分区

1.2 当前批处理过程中的挑战

这个过程有效，但也有其自身的缺陷：

1. 时间和成本——每天都需要覆盖整个历史表
2. 数据版本控制——没有开箱即用的数据和清单版本控制（回滚、并发读取和写入、时间点查询、时间旅行以及相关功能不存在）
3. 写入放大——日常历史数据覆盖场景中的外部（或自我管理）数据版本控制增加了写入放大，从而占用更多的 S3 存储

借助Apache Hudi，我们希望在将数据摄取到数据湖中的同时，找到更好的重复数据删除和数据版本控制优化解决方案。

2. Hudi 数据湖 — 查询模式

当我们开始在我们的数据湖上实现 Apache Hudi 的旅程时，我们根据表的主要用户的查询模式将表分为 2 类。

• 面向ETL ：这是指我们从各种生产系统摄取到数据湖中的大多数原始/基本快照表。 如果这些表被 ETL 作业广泛使用，那么我们将每日数据分区保持在 updated_date，这样下游作业可以简单地读取最新的 updated_at 分区并（重新）处理数据。
• 面向分析师：通常包括维度表和业务分析师查询的大部分计算 OLAP，分析师通常需要查看基于事务（或事件）created_date 的数据，而不太关心 updated_date。

这是一个示例电子商务订单数据流，从摄取到数据湖到创建 OLAP，最后到业务分析师查询它

由于两种类型的表的日期分区列不同，我们采用不同的策略来解决这两个用例。

2.1 面向分析师的表/OLAP（按 created_date 分区）

在 Hudi 中，我们需要指定分区列和主键列，以便 Hudi 可以为我们处理更新和删除。
以下是我们如何处理面向分析师的表中的更新和删除的逻辑：

• 读取上游数据的 D-n 个 updated_date 分区。
• 应用数据转换。 现在这个数据将只有新的插入和很少的更新记录。
• 发出 hudi upsert 操作，将处理后的数据 upsert 到目标 Hudi 表。

由于主键和 created_date 对于退出和传入记录保持相同，Hudi 通过使用来自传入记录 created_date 和 primary_key 列的此信息获取现有记录的分区和分区文件路径。

2.2 面向ETL（按更新日期分区）

当我们开始使用 Hudi 时，在阅读了许多博客和文档之后，在 created_date 上对面向 ETL 的表进行分区似乎是合乎逻辑的。
此外 Hudi 提供增量消费功能，允许我们在 created_date 上对表进行分区，并仅获取在 D-1 或 D-n 上插入（插入或更新）的那些记录。

1. “created_date”分区的挑战

这种方法在理论上效果很好，但在改造传统的日常批处理过程中的增量消费时，它带来了其他一系列挑战：
Hudi 维护了在不同时刻在表上执行的所有操作的时间表，这些提交包含有关作为 upsert 的一部分插入或重写的部分文件的信息，我们将此 Hudi 表称为 Commit Timeline。
这里要注意的重要信息是增量查询基于提交时间线，而不依赖于数据记录中存在的实际更新/创建日期信息。

• 冷启动：当我们将现有的上游表迁移到 Hudi 时，D-1 Hudi 增量查询将获取完整的表，而不仅仅是 D-1 更新。发生这种情况是因为在开始时，整个表是通过在 D-1 提交时间线内发生的单个初始提交或多个提交创建的，并且缺少真正的增量提交信息。
• 历史数据重新摄取：在每个常规增量 D-1 拉取中，我们期望仅在 D-1 上更新的记录作为输出。但是在重新摄取历史数据的情况下，会再次出现类似于前面描述的冷启动问题的问题，并且下游作业也会出现 OOM。

历史数据重新摄取：在每个常规增量 D-1 拉取中，我们期望仅在 D-1 上更新的记录作为输出。但是在重新摄取历史数据的情况下，会再次出现类似于前面描述的冷启动问题的问题，并且下游作业也会出现 OOM。

作为面向 ETL 的作业的解决方法，我们尝试将数据分区保持在 updated_date 本身，然而这种方法也有其自身的挑战。

2. “updated_date”分区的挑战

我们知道 Hudi 表的本地索引，Hudi 依靠索引来获取存储在数据分区本地目录中的 Row-to-Part_file 映射。因此，如果我们的表在 updated_date 进行分区，Hudi 无法跨分区自动删除重复记录。
Hudi 的全局索引策略要求我们保留一个内部或外部索引来维护跨分区的数据去重。对于大数据量，每天大约 2 亿条记录，这种方法要么运行缓慢，要么因 OOM 而失败。
因此，为了解决更新日期分区的数据重复挑战，我们提出了一种全新的重复数据删除策略，该策略也具有很高的性能。

3. “新”重复数据删除策略

• 查找更新 - 从每日增量负载中，仅过滤掉更新（1-10% 的 DI 数据）（其中 updated_date> created_date）（快速，仅映射操作）
• 找到过时更新 - 将这些“更新”与下游 Hudi 基表广播连接。 由于我们只获取更新的记录（仅占每日增量的 1-10%），因此可以实现高性能的广播连接。 这为我们提供了与更新记录相对应的基础 Hudi 表中的所有现有记录
• 删除过时更新——在基本 Hudi 表路径上的这些“过时更新”上发出 Hudi 删除命令
• 插入 - 在基本 hudi 表路径上的完整每日增量负载上发出 hudi insert 命令

进一步优化用 true 填充陈旧更新中的 _hoodie_is_deleted 列，并将其与每日增量负载结合。 通过基本 hudi 表路径发出此数据的 upsert 命令。 它将在单个操作（和单个提交）中执行插入和删除。

4. Apache Hudi 的优势

• 时间和成本——Hudi 在重复数据删除时不会覆盖整个表。 它只是重写接收更新的部分文件。 因此较小的 upsert 工作
• 数据版本控制——Hudi 保留表版本（提交历史），因此提供实时查询（时间旅行）和表版本回滚功能。
• 写入放大——由于只有部分文件被更改并保留用于数据清单版本控制，我们不需要保留完整数据的版本。 因此整体写入放大是最小的。

作为数据版本控制的另一个好处，它解决了并发读取和写入问题，因为数据版本控制使并发读取器可以读取数据文件的版本控制副本，并且当并发写入器用新数据覆盖同一分区时不会抛出 FileNotFoundException 文件。