「Apache Hudi系列」核心概念与架构设计总结

简介

Apache Hudi依赖 HDFS 做底层的存储，所以可以支撑非常大规模的数据存储。同时基于下面两个原语，Hudi可以解决流批一体的存储问题。

提供了在hadoop兼容的存储之上存储大量数据，同时它还提供两种原语：

• Update/Delete 记录：Hudi 支持更新/删除记录，使用文件/记录级别索引，同时对写操作提供事务保证。查询可获取最新提交的快照来产生结果。

• 变更流：支持增量获取表中所有更新/插入/删除的记录，从指定时间点开始进行增量查询，可以实现类似 Kafka 的增量消费机制。

一些设计原则

流式读/写：Hudi借鉴了数据库设计的原理，从零设计，应用于大型数据集记录流的输入和输出。为此，Hudi提供了索引实现，可以将记录的键快速映射到其所在的文件位置。同样，对于流式输出数据，Hudi通过其特殊列添加并跟踪记录级的元数据，从而可以提供所有发生变更的精确增量流。

自管理：Hudi注意到用户可能对数据新鲜度（写友好）与查询性能（读/查询友好）有不同的期望，它支持了三种查询类型，这些类型提供实时快照，增量流以及稍早的纯列数据。在每一步，Hudi都努力做到自我管理（例如自动优化编写程序的并行性，保持文件大小）和自我修复（例如：自动回滚失败的提交），即使这样做会稍微增加运行时成本（例如：在内存中缓存输入数据已分析工作负载）。如果没有这些内置的操作杠杆/自我管理功能，这些大型流水线的运营成本通常会翻倍。

万物皆日志：Hudi还具有 append only、云数据友好的设计，该设计实现了日志结构化存储系统的原理，可以无缝管理所有云提供商的数据。

键-值数据模型：在写方面，Hudi表被建模为键值对数据集，其中每条记录都有一个唯一的记录键。此外，一个记录键还可以包括分区路径，在该路径下，可以对记录进行分区和存储。这通常有助于减少索引查询的搜索空间。

Hudi表设计

Hudi表的三个主要组件：

1. 有序的时间轴元数据:类似于数据库事务日志。
2. 分层布局的数据文件:实际写入表中的数据。
3. 索引(多种实现方式):映射包含指定记录的数据集。

另外，针对数据的写入和查询，Hudi提供一些非常重要的功能例如upsert、mvvc等。

时间轴TimeLine

Timeline 是 HUDI 用来管理提交（commit）的抽象，每个 commit 都绑定一个固定时间戳，分散到时间线上。在 Timeline 上，每个 commit 被抽象为一个 HoodieInstant，一个 instant 记录了一次提交 (commit) 的行为、时间戳、和状态。HUDI 的读写 API 通过 Timeline 的接口可以方便的在 commits 上进行条件筛选，对 history 和 on-going 的 commits 应用各种策略，快速筛选出需要操作的目标 commit。

如图所示：

Hudi维护了一条包含在不同的即时时间(instant time)对数据集做的所有instant操作的timeline，从而提供表的即时视图，同时还有效支持按到达顺序进行数据检索。时间轴类似于数据库的redo/transaction日志，由一组时间轴实例组成。Hudi保证在时间轴上执行的操作的原子性和基于即时时间的时间轴一致性。时间轴被实现为表基础路径下.hoodie元数据文件夹下的一组文件。具体来说，最新的instant被保存为单个文件，而较旧的instant被存档到时间轴归档文件夹中，以限制writers和queries列出的文件数量。

一个Hudi 时间轴instant由下面几个组件构成：

1. 操作类型：对数据集执行的操作类型;
2. 即时时间：即时时间通常是一个时间戳(例如：20190117010349)，该时间戳按操作开始时间的顺序单调增加;
3. 即时状态：instant的当前状态;每个instant都有avro或者json格式的元数据信息，详细的描述了该操作的状态以及这个即时时刻instant的状态。

关键的Instant操作类型有：

1. COMMIT：一次提交表示将一组记录原子写入到数据集中;
2. CLEAN: 删除数据集中不再需要的旧文件版本的后台活动;
3. DELTA_COMMIT:将一批记录原子写入到MergeOnRead存储类型的数据集中，其中一些/所有数据都可以只写到增量日志中;
4. COMPACTION: 协调Hudi中差异数据结构的后台活动，例如：将更新从基于行的日志文件变成列格式。在内部，压缩表现为时间轴上的特殊提交;
5. ROLLBACK: 表示提交/增量提交不成功且已回滚，删除在写入过程中产生的所有部分文件;
6. SAVEPOINT: 将某些文件组标记为"已保存"，以便清理程序不会将其删除。在发生灾难/数据恢复的情况下，它有助于将数据集还原到时间轴上的某个点;

任何给定的即时都会处于以下状态之一:

• REQUESTED:表示已调度但尚未初始化;
• INFLIGHT: 表示当前正在执行该操作;
• COMPLETED: 表示在时间轴上完成了该操作.

数据文件

Hudi将表组织成DFS上基本路径下的文件夹结构中。如果表是分区的，则在基本路径下还会有其他的分区，这些分区是包含该分区数据的文件夹，与Hive表非常类似。每个分区均由相对于基本路径的分区路径唯一标识。在每个分区内，文件被组织成文件组，由文件ID唯一标识。其中每个切片包含在某个提交/压缩即时时间生成的基本列文件（.parquet）以及一组日志文件（.log*），该文件包含自生成基本文件以来对基本文件的插入/更新。Hudi采用了MVCC设计，压缩操作会将日志和基本文件合并以产生新的文件片，而清理操作则将未使用的/较旧的文件片删除以回收HDFS上的空间。

下图展示了一个分区内的文件结构：

文件版本

一个新的 base commit time 对应一个新的 FileSlice，实际就是一个新的数据版本。HUDI 通过 TableFileSystemView 抽象来管理 table 对应的文件，比如找到所有最新版本 FileSlice 中的 base file （Copy On Write Snapshot 读）或者 base + log files（Merge On Read 读)。通过 Timeline 和 TableFileSystemView 抽象，HUDI 实现了非常便捷和高效的表文件查找。

文件格式

Hoodie 的每个 FileSlice 中包含一个 base file （merge on read 模式可能没有）和多个 log file （copy on write 模式没有）。

每个文件的文件名都带有其归属的 FileID（即 FileGroup Identifier）和 base commit time（即 InstanceTime）。通过文件名的 group id 组织 FileGroup 的 logical 关系；通过文件名的 base commit time 组织 FileSlice 的逻辑关系。

HUDI 的 base file (parquet 文件) 在 footer 的 meta 去记录了 record key 组成的 BloomFilter，用于在 file based index 的实现中实现高效率的 key contains 检测。只有不在 BloomFilter 的 key 才需要扫描整个文件消灭假阳。

HUDI 的 log （avro 文件）是自己编码的，通过积攒数据 buffer 以 LogBlock 为单位写出，每个 LogBlock 包含 magic number、size、content、footer 等信息，用于数据读、校验和过滤。

索引设计

Hudi通过索引机制提供高效的upsert操作，该机制会将一个记录键+分区路径组合一致性的映射到一个文件ID.这个记录键和文件组/文件ID之间的映射自记录被写入文件组开始就不会再改变。简而言之，这个映射文件组包含了一组文件的所有版本。Hudi当前提供了3种索引实现(HBaseIndex、HoodieBloomIndex(HoodieGlobalBloomIndex)、InMemoryHashIndex)来映射一个记录键到包含该记录的文件ID。这将使我们无需扫描表中的每条记录，就可显著提高upsert速度。

Hudi索引可以根据其查询分区记录的能力进行分类：

1. 全局索引：不需要分区信息即可查询记录键映射的文件ID。比如，写程序可以传入null或者任何字符串作为分区路径（partitionPath）,但索引仍然会查找到该记录的位置。全局索引在记录键在整张表中保证唯一的情况下非常有用，但是查询的消耗随着表的大小呈函数式增加。

2. 非全局索引：与全局索引不同，非全局索引依赖分区路径(partitionPath),对于给定的记录键，它只会在给定分区路径下查找该记录。这比较适合总是同时生成分区路径和记录键的场景，同时还能享受到更好的扩展性，因为查询索引的消耗只与写入到该分区下数据集大小有关系。

表类型

Copy On Write

COW表写的时候数据直接写入basefile,（parquet）不写log文件。所以COW表的文件片只包含basefile(一个parquet文件构成一个文件片)。这种的存储方式的Spark DAG相对简单。关键目标是是使用partitioner将tagged Hudi记录RDD（所谓的tagged是指已经通过索引查询，标记每条输入记录在表中的位置）分成一些列的updates和inserts.为了维护文件大小，我们先对输入进行采样，获得一个工作负载profile,这个profile记录了输入记录的insert和update、以及在分区中的分布等信息。把数据从新打包，这样:

• 对于updates,该文件ID的最新版本都将被重写一次，并对所有已更改的记录使用新值。
• 对于inserts,记录首先打包到每个分区路径中的最小文件中，直到达到配置的最大大小。之后的所有剩余记录将再次打包到新的文件组，新的文件组也会满足最大文件大小要求。

Copy On Write 类型表每次写入都会生成一个新的持有base file（对应写入的 instant time）的 FileSlice。

用户在snapshot读取的时候会扫描所有最新的FileSlice下的base file。

Merge On Read

MOR表写数据时，记录首先会被快速的写进日志文件，稍后会使用时间轴上的压缩操作将其与基础文件合并。根据查询是读取日志中的合并快照流还是变更流，还是仅读取未合并的基础文件，MOR表支持多种查询类型。在高层次上，MOR writer在读取数据时会经历与COW writer 相同的阶段。这些更新将追加到最新文件篇的最新日志文件中，而不会合并。对于insert,Hudi支持两种模式：

1. 插入到日志文件：有可索引日志文件的表会执行此操作(HBase索引);
2. 插入parquet文件：没有索引文件的表(例如布隆索引)

与写时复制（COW）一样，对已标记位置的输入记录进行分区，将所有发往相同文件id的upsert分到一组。这批upsert会作为一个或多个日志块写入日志文件。Hudi允许客户端控制日志文件大小。对于写时复制(COW)和读时合并(MOR)writer来说，Hudi的WriteClient是相同的。几轮数据的写入将会累积一个或多个日志文件。这些日志文件与基本的parquet文件(如果有)一起构成一个文件片，而这个文件片代表该文件的一个完整版本。

这种表是用途最广、最高级的表。为写(可以指定不同的压缩策略，吸收突发写流量)和查询(例如权衡数据的时效性和查询性能)提供了很大的灵活性。

Merge On Read 表的写入行为，依据 index 的不同会有细微的差别：

• 对于 BloomFilter 这种无法对 log file 生成 index 的索引方案，对于 INSERT 消息仍然会写 base file （parquet format），只有 UPDATE 消息会 append log 文件（因为 base file 已经记录了该 UPDATE 消息的 FileGroup ID）。
• 对于可以对 log file 生成 index 的索引方案，例如 Flink writer 中基于 state 的索引，每次写入都是 log format，并且会不断追加和 roll over。

Merge On Read 表的读在 READ OPTIMIZED 模式下，只会读最近的经过 compaction 的 commit。

数据读写流程

读流程

Snapshot读

读取所有 partiiton 下每个 FileGroup 最新的 FileSlice 中的文件，Copy On Write 表读 parquet 文件，Merge On Read 表读 parquet + log 文件

Incremantal读

根据https://hudi.apache.org/docs/querying_data.html#spark-incr-query描述，当前的 Spark data source 可以指定消费的起始和结束 commit 时间，读取 commit 增量的数据集。但是内部的实现不够高效：拉取每个 commit 的全部目标文件再按照系统字段 hoodie_commit_time apply 过滤条件。

Streaming 读

HUDI Flink writer 支持实时的增量订阅，可用于同步 CDC 数据，日常的数据同步 ETL pipeline。Flink 的 streaming 读做到了真正的流式读取，source 定期监控新增的改动文件，将读取任务下派给读 task。

写流程

写操作

• UPSERT：默认行为，数据先通过 index 打标(INSERT/UPDATE)，有一些启发式算法决定消息的组织以优化文件的大小 => CDC 导入
• INSERT：跳过 index，写入效率更高 => Log Deduplication
• BULK_INSERT：写排序，对大数据量的 Hudi 表初始化友好，对文件大小的限制 best effort（写 HFile）

写流程(UPSERT)

Copy On Write

• 先对 records 按照 record key 去重
• 首先对这批数据创建索引 (HoodieKey => HoodieRecordLocation)；通过索引区分哪些 records 是 update，哪些 records 是 insert（key 第一次写入）
• 对于 update 消息，会直接找到对应 key 所在的最新 FileSlice 的 base 文件，并做 merge 后写新的 base file (新的 FileSlice)
• 对于 insert 消息，会扫描当前 partition 的所有 SmallFile（小于一定大小的 base file），然后 merge 写新的 FileSlice；如果没有 SmallFile，直接写新的 FileGroup + FileSlice

Merge On Read

• 先对 records 按照 record key 去重（可选）
• 首先对这批数据创建索引 (HoodieKey => HoodieRecordLocation)；通过索引区分哪些 records 是 update，哪些 records 是 insert（key 第一次写入）
• 如果是 insert 消息，如果 log file 不可建索引（默认），会尝试 merge 分区内最小的 base file （不包含 log file 的 FileSlice），生成新的 FileSlice；如果没有 base file 就新写一个 FileGroup + FileSlice + base file；如果 log file 可建索引，尝试 append 小的 log file，如果没有就新写一个 FileGroup + FileSlice + base file
• 如果是 update 消息，写对应的 file group + file slice，直接 append 最新的 log file（如果碰巧是当前最小的小文件，会 merge base file，生成新的 file slice）log file 大小达到阈值会 roll over 一个新的

写流程(INSERT)

Copy On Write

• 先对 records 按照 record key 去重（可选）
• 不会创建 Index
• 如果有小的 base file 文件，merge base file，生成新的 FileSlice + base file，否则直接写新的 FileSlice + base file

Merge On Read

• 先对 records 按照 record key 去重（可选）
• 不会创建 Index
• 如果 log file 可索引，并且有小的 FileSlice，尝试追加或写最新的 log file；如果 log file 不可索引，写一个新的 FileSlice + base file

总结

主要是我个人收集和翻阅Hudi社区的一些资料过程中的总结。目前Hudi版本到了0.11版本。细节上可能有所差异，以社区为准。