数据湖之Hudi基础：核心原理

本文重点参考自官方文档（v0.12.1英文好的建议直接看官方文档：https://hudi.apache.org/docs/0.12.1/concepts/

基本概念

时间轴TimeLine

Hudi的核心是维护表上在不同的即时时间（instants执行的所有操作的时间轴（timeline

一个instant由以下三个部分组成：

1.Instants action 在表上执行的操作类型

Instants action的值取类型如下

• COMMITS：一次Commit表示将一批数据原子性的写入一个表
• CLEANS：清除表不再需要的旧版本文件的后台活动
• DELTA_COMMIT：增量提交指的是将一批数据原子性地写入一个MergeOnRead类型地表其中部分或所有数据可以写入增量日志。
• COMPACTION：合并Hudi内部差异数据结构地后台活动例如：将更新操作从基于行地log日志文件合并到列式存储地数据文件。在内部COMPACTION体现为timeline上地特殊提交
• ROLLBACK：表示当commit、delta_commit不成功时进行回滚其会删除在写入过程中产生地部分文件
• SAVEPOINT：将某些文件组标记为已保存以便其不会被删除。发生问题时可以根据SAVEPOINT将数据集还原到时间轴上地某个点。

2.Instant time

一般是一个时间戳比如：20230117121314098按照动作开始时间的顺序单调增加

3.State

状态

• REQUESTED：表示某个action已经调度但是尚未执行
• INFLIGHT：表示action当前正在执行
• COMPLETED：表示timeline上的action已经完成

两个时间概念

hudi的两个重要时间概念

• Arrival time：数据到达Hudi的时间的commit time（类比flink的摄入时间
• Event time：数据自己的时间

那么当一条延迟的数据到来是否可以被消费到呢？他会落盘到哪个分区？

上图中采用时间（小时作为分区字段从 10:00 开始陆续产生各种 commits10:20 来了一条 9:00 的数据根据event time该数据仍然可以落到 9:00 对应的分区通过 timeline 直接消费 10:00 （commit time之后的增量更新（只消费有新 commits 的 group那么这条延迟的数据仍然可以被消费到。

文件布局（FileLayout

Hudi的一个表文件结构如何？

Hudi存储分为两部分：

• 元数据：hoodie目录对应着表的元数据信息包括表的版本管理（Timeline、归档目录（存放过时的instant一个instant记录了一次提交的行为、时间戳和状态（action、time、stateHudi以时间轴的形式维护了在数据集上执行的所有操作的元数据
• 数据：和Hive一样以分区方式存放数据；分区里有BaseFile（.parquet和LogFile（.log.*

文件管理如下

• 数据表组织成分布式文件系统基本路径（basepath下的目录结构

• 表被划分为多个分区这些分区时包含该分区数据的文件夹非常类似Hive表

• 在每个分区中文件被组织成文件组由文件ID唯一标识

• 每个文件组包含几个文件分片 FileSlice

• 每个文件分片包含：

  • 一个基本文件（.parquet：在某个commit、compaction即时时间（instant time生成的（Merge On Read可能没有
  • 多个日志文件（.log.*：这些日志文件包含自生成基本文件以来对基本文件的插入、更新（COW没有

• 多版本并发控制文件（Multiversion Concurrency ControlMVCC

  • compaction操作：合并日志和基本文件以及产生新的文件片
  • clean操作：清除不适用的旧的文件片以回收文件系统上的空间

  

• basefile（parquet文件在footer的meta去记录了record key组成的BloomFilter用于在file based index的视线中实现高效率的key contants检测。只有不在BloomFilter的key才需要扫描整个文件排除BloomFilter的假阳

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

  

索引 Index

• 1原理

  Hudi通过索引机制提供高效的upserts具体是将给定的hoodie key(record key + partition path)与文件id（文件组建立唯一映射。这种映射关系数据第一次写入文件后保持不变所以一个 FileGroup 包含了一批 record 的所有版本记录。Index 用于区分消息是 INSERT 还是 UPDATE。

​

​ 为了避免不必要的读写引入索引。那么在更新数据时可以快速被定位到对应FileGroup。上图白色为基本文件黄色是更新数据。有了索引机制可以做到：避免读取不需要的文件、避免更新不必要的文件、无需将更新数据与历史数据做分布式关联只需要在 File Group 内做合并。

• 2索引选项

Index类型	原理	优点	缺点
Bloom Index	默认配置使用布隆过滤器来判断记录存在与否也可选使用record key的范围裁剪需要的文件	效率高不依赖外部系统数据和索引保持一致性	因假阳性问题还需回溯原文件再查找一遍
Simple Index	把update/delete操作的新数据和老数据进行join	实现最简单无需额外的资源	性能比较差
HBase Index	把index存放在HBase里面。在插入 File Group定位阶段所有task向HBase发送 Batch Get 请求获取 Record Key 的 Mapping 信息	对于小批次的keys查询效率高	需要外部的系统增加了运维压力
Flink State-based Index	HUDI 在 0.8.0 版本中实现的 Flink witer采用了 Flink 的 state 作为底层的 index 存储每个 records 在写入之前都会先计算目标 bucket ID。	不同于 BloomFilter Index避免了每次重复的文件 index 查找

注意：Flink只有一种state based index（和bucket_index其他index是Spark可选配置。

• 3全局索引和非全局索引

  • 全局索引：全表范围要求键唯一保证有且只有一个对应记录。但是表增大后update/delete操作损失性能越高因此适用于小表

  • 非全局索引：默认索引。保证数据在分区内唯一。依靠写入器为同一个记录的update/delete提供一致的分区路径同时大幅提高了效率更适用于大表。

​ 从index的维护成本和写入性能的角度考虑维护一个global index的难度更大对写入性能的影响也更大所以需要non-global index。

当使用HBase 作索引时相当于一个全局索引

当使用bloom和simple index时分别有全局选项

hoodie.index.type=GLOBAL_BLOOM

hoodie.index.type=GLOBAL_SIMPLE

• 4索引选择策略

  • 对事实表的延迟更新

    ​ 许多公司会在NoSQL数据存储中存放大量的交易数据。例如共享出行的行程表、股票买卖记录的表、和电商的订单表。这些表通常一直在增长且大部分的更新随机发生在较新的记录上而对旧记录有着长尾分布型的更新。这通常是源于交易关闭或者数据更正的延迟性。换句话说大部分更新会发生在最新的几个分区上而小部分会在旧的分区。

  

  ​ 对于这样的作业模式布隆索引就能表现地很好因为查询索引可以靠设置得当的布隆过滤器来裁剪很多数据文件。另外如果生成的键可以以某种顺序排列参与比较的文件数会进一步通过范围裁剪而减少。Hudi用所有文件的键域来构造区间树这样能来高效地依据输入的更删记录的键域来排除不匹配的文件。

  ​ 为了高效地把记录键和布隆过滤器进行比对即尽量减少过滤器的读取和均衡执行器间的工作量Hudi缓存了输入记录并使用了自定义分区器和统计规律来解决数据的偏斜。有时如果布隆过滤器的假阳性率过高查询会增加数据的打乱操作。Hudi支持动态布隆过滤器（设置hoodie.bloom.index.filter.type=DYNAMIC_V0。它可以根据文件里存放的记录数量来调整大小从而达到设定的假阳性率。

  • 对事件表的去重

    ​ 事件流无处不在事件数通常是事实表大小的10-100倍

    ​ 事件通常把时间（到达时间、处理时间作为首类处理对象比如物联网的事件流、点击流数据、广告曝光数等等。由于这些大部分都是仅追加的数据插入和更新只存在于最新的几个分区中。由于重复事件可能发生在整个数据管道的任一节点在存放到数据湖前去重是一个常见的需求。

    

    ​ 去重时虽然可以用索引但索引存储的消耗也会随着事件增长而线性增长以至于变得不可行。

    ​ 事实上有范围裁剪功能的布隆索引是最佳的解决方案。我们可以利用作为首类处理对象的时间来构造由事件时间戳和事件id（event_ts+event_id)组成的键这样插入的记录就有了单调增长的键。这会在最新的几个分区里大幅提高裁剪文件的效益。

  • 对维度表的随机更新删除

    

    ​ 如果范围比较不能裁剪更多文件那bloom索引并不能带来较好效益。在这种随机写入的场景更新操作通常会触及表里大多数文件从而导致bloom功能退化（对所有文件都可能标阳导致所有文件都要scan此时使用简单索引更合适因为它不用提前剪裁而是直接和所有文件的所需字段连接如果有能力也可以使用HBase索引它对随机场景会有更好的查询效率。

    ​ 当使用全局索引时也可以考虑通过设置hoodie.bloom.index.update.partition.path=true或hoodie.simple.index.update.partition.path=true来处理 的情况；例如对于以所在城市分区的用户表会有用户迁至另一座城市的情况。这些表也非常适合采用Merge-On-Read表型。

表类型（Table Types

Copy On Write

简称COW表只有数据文件/基本文件（.parquet,没有增量日志文件（.log.*

对每一个新批次写入都将创建相应数据文件的新版本（新的FileSlice新版本文件包括旧版本文件的记录以及来自传入批次的记录（全量最新。

​ 假设我们有 3 个文件组（如上图。我们进行一批新的写入在索引后我们发现这些记录与File group 1 和File group 2 匹配然后有新的插入我们将为其创建一个新的文件组（File group 4。

​ 因此data_file1 和 data_file2 都将创建更新的版本data_file1 V2 是data_file1 V1 的内容与data_file1 中传入批次匹配记录的记录合并。

​ 由于在写入期间进行合并COW 会产生一些写入延迟。但是COW 的优势在于它的简单性不需要其他表服务（如压缩也相对容易调试。

Merge On Read

简称MOR表。包含列存的基本文件（.parquet和行存的增量日志文件（基于行的avro格式.log.*。

​ 顾名思义MOR表的合并成本在读取端。因此在写入期间我们不会合并或创建较新的数据文件版本。标记/索引完成后对于具有要更新记录的现有数据文件Hudi 创建增量日志文件并适当命名它们以便它们都属于一个文件组。

​ 读取端将实时合并基本文件及其各自的增量日志文件。每次的读取延迟都比较高（因为查询时进行合并所以 Hudi 使用压缩机制来将数据文件和日志文件合并在一起并创建更新版本的数据文件。

​ 用户可以选择内联或异步模式运行压缩。Hudi也提供了不同的压缩策略供用户选择最常用的一种是基于提交的数量。例如可以将压缩的最大增量日志配置为 4。这意味着在进行 4 次增量写入后将对数据文件进行压缩并创建更新版本的数据文件。压缩完成后读取端只需要读取最新的数据文件而不必关心旧版本文件。

MOR表的写入行为依据 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。

MOR和COW对比

	CopyOnWrite	MergeOnRead
数据延迟	高	低
查询延迟	低	高
Update(I/O) 更新成本	高（重写整个Parquet文件	低（追加到增量日志
Parquet文件大小	低（更新成本I/O高	较大（低更新成本
写放大	大	低（取决于压缩策略

查询类型

hudi支持如下三种查询类型

• Snapshot Queries

  ​ 快照查询可查指定commit/delta commit即时操作后表的最新快照。

  ​ 在读时合并（MOR表的情况下它通过即时合并最新文件片的基本文件和增量文件来提供近实时表（几分钟。

  ​ 对于写时复制（COW它可以替代现有的parquet表（或相同基本文件类型的表同时提供upsert/delete和其他写入方面的功能可以理解为查询最新版本的Parquet数据文件。

  下图是COW的快照查询：

  

• Incremental Queres

  增量查询可以查询给定commit/delta commit即时操作以来新写入的数据。有效的提供变更流来启用增量数据管道。

• Read Optimized Queries

  读优化查询可查看给定的commit/compact即时操作的表的最新快照。仅将最新文件片的基本/列文件暴露给查询并保证与非Hudi表相同的列查询性能。

  下图是MOR表的快照查询与读优化查询的对比：

​

Read Optimized Queries是对Merge On Read表类型快照查询的优化。

	Snapshot	Read Optimized
数据延迟	低	高
查询延迟	高（合并列式基础文件+行式增量日志文件	低(原始列式基础文件)

• 不同表支持的查询类型

  Table Type	支持的查询类型
  COW	Snapshot queries+ Incremental queries
  MOR	Snapshot queries+ Incremental queries + Read Optimized Queries

数据写

写操作

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

写流程（UPSERT

COW

• （1先对 records 按照 record key 去重

• （2首先对这批数据创建索引 (HoodieKey => HoodieRecordLocation)；通过索引区分哪些 records 是 update哪些 records 是 insert（key 第一次写入

• （3对于 update 消息会直接找到对应 key 所在的最新 FileSlice 的 base 文件并做 merge 后写新的 base file (新的 FileSlice)

• （4对于 insert 消息会扫描当前 partition 的所有 SmallFile（小于一定大小的 base file然后 merge 写新的 FileSlice；如果没有 SmallFile直接写新的 FileGroup + FileSlice

MOR

• （1先对 records 按照 record key 去重（可选

• （2首先对这批数据创建索引 (HoodieKey => HoodieRecordLocation)；通过索引区分哪些 records 是 update哪些 records 是 insert（key 第一次写入

• （3如果是 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

• （4如果是 update 消息写对应的 file group + file slice直接 append 最新的 log file（如果碰巧是当前最小的小文件会 merge base file生成新的 file slice

• （5log file 大小达到阈值会 roll over 一个新的

写流程（INSERT

COW

• （1先对 records 按照 record key 去重（可选

• （2不会创建 Index

• （3如果有小的 base file 文件merge base file生成新的 FileSlice + base file否则直接写新的 FileSlice + base file

MOR

• （1先对 records 按照 record key 去重（可选

• （2不会创建 Index

• （3如果 log file 可索引并且有小的 FileSlice尝试追加或写最新的 log file；如果 log file 不可索引写一个新的 FileSlice + base file

写流程（INSERT OVERWRITE

在同一分区中创建新的文件组集。现有文件组被标记为 “删除”。根据新记录的数量创建新的文件组

COW

在插入分区之前	插入相同数量的记录覆盖	插入覆盖更多的记录	插入重写1条记录
分区包含file1-t0.parquetfile2-t0.parquet。	分区将添加file3-t1.parquetfile4-t1.parquet。file1, file2在t1后的元数据中被标记为无效。	分区将添加file3-t1.parquetfile4-t1.parquetfile5-t1.parquet…fileN-t1.parquet。file1, file2在t1后的元数据中被标记为无效	分区将添加file3-t1.parquet。file1, file2在t1后的元数据中被标记为无效。

MOR

在插入分区之前	插入相同数量的记录覆盖	插入覆盖更多的记录	插入重写1条记录
分区包含file1-t0.parquetfile2-t0.parquet。.file1-t00.log	file3-t1.parquetfile4-t1.parquet。file1, file2在t1后的元数据中被标记为无效。	file3-t1.parquet, file4-t1.parquet…fileN-t1.parquetfile1, file2在t1后的元数据中被标记为无效	分区将添加file3-t1.parquet。file1, file2在t1后的元数据中被标记为无效。

优点

• （1COW和MOR在执行方面非常相似。不干扰MOR的compaction。

• （2减少parquet文件大小。

• （3不需要更新关键路径中的外部索引。索引实现可以检查文件组是否无效（类似于在HBaseIndex中检查commit是否无效的方式。

• （4可以扩展清理策略在一定的时间窗口后删除旧文件组。

缺点

• （1需要转发以前提交的元数据。

  • 在t1比如file1被标记为无效我们在t1.commit中存储 “invalidFiles=file1”(或者在MOR中存储deltacommit)
  • 在t2比如file2也被标记为无效。我们转发之前的文件并在t2.commit中标记 “invalidFiles=file1, file2”（或MOR的deltacommit

• （2忽略磁盘中存在的parquet文件也是Hudi的一个新行为, 可能容易出错,我们必须认识到新的行为并更新文件系统的所有视图来忽略它们。这一点可能会在实现其他功能时造成问题。

Key 生成策略

用来生成 HoodieKey（record key + partition path目前支持以下策略：

• 支持多个字段组合 record keys

• 支持多个字段组合的 parition path （可定制时间格式Hive style path name

• 非分区表

删除策略

• 1逻辑删：将 value 字段全部标记为 null

• 2物理删：

  • （1通过 OPERATION_OPT_KEY 删除所有的输入记录

  • （2配置 PAYLOAD_CLASS_OPT_KEY = org.apache.hudi.EmptyHoodieRecordPayload 删除所有的输入记录

  • （3在输入记录添加字段：_hoodie_is_deleted

Hudi写小结

通过对写流程的梳理可以了解到 Apache Hudi 相对于其他数据湖方案的核心优势：

• （1写入过程充分优化了文件存储的小文件问题Copy On Write 写会一直将一个 bucket （FileGroup的 base 文件写到设定的阈值大小才会划分新的 bucket；Merge On Read 写在同一个 bucket 中log file 也是一直 append 直到大小超过设定的阈值 roll over。

• （2对 UPDATE 和 DELETE 的支持非常高效一条 record 的整个生命周期操作都发生在同一个 bucket不仅减少小文件数量也提升了数据读取的效率（不必要的 join 和 merge。

数据读

Snapshot读

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

Incremental读

​ Spark读：https://hudi.apache.org/docs/querying_data.html#spark-incr-query

​ 当前的 Spark data source 可以指定消费的起始和结束 commit 时间读取 commit 增量的数据集。但是内部的实现不够高效：拉取每个 commit 的全部目标文件再按照系统字段 hoodie_commit_time apply 过滤条件。

Streaming读

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

Compaction

• （1没有 base file：走 copy on write insert 流程直接 merge 所有的 log file 并写 base file

• （2有 base file：走 copy on write upsert 流程先读 log file 建 index再读 base file最后读 log file 写新的 base file

Flink 和 Spark streaming 的 writer 都可以 apply 异步的 compaction 策略按照间隔 commits 数或者时间来触发 compaction 任务在独立的 pipeline 中执行。