1、数据湖定义

业界对于数据湖的定义存在一定争议，个人认为数据湖就是针对传统hive数仓不支持acid、upsert、schema evolution等痛点上，提出的一种数据存储库。
hive的痛点：hive主要特性是提供了sql解析和元数据管理的功能，统一管理了存储在hdfs上数据的shcmea信息。但是设计之初hive并没有考虑支持upsert，schema evolution等特性，基于这些业务痛点，数据湖应运而生。

2、湖仓湖的核心特点

2.1、ACID特性的支持

ACDID即数据库中事务特性，但是数据湖的事务和oltp数据库的数据特性不同，数据湖是粗粒度的事务控制(filegroup级别)。所谓事务，本质就是一个并发问题，本质在于解决读写冲突和写写冲突。以hudi为例，支持行级并发控制和列级别并发控制(通过TimeLine实现)。

并发控制：hudi中的每个commit都被抽象为TimeLine上的一个instance，instance记录了本次操作的行为、时间戳和状态。

Instant action（即时动作）

在表上执行的动作类型；Hudi保证在时间轴上执行的操作是原子的，并且是基于即时时间的时间轴一致的。

COMMITS：表示将一批记录原子地写入表。

CLEANS：清除表中不再需要的旧版本文件的后台活动。

DELTA_COMMIT：增量提交是指将一批记录原子地写入MergeOnRead类型的表，其中一些/所有数据可以直接写入增量日志。

COMPACTION ：协调Hudi中不同数据结构的后台活动，例如:将更新从基于行的日志文件移动到柱状格式。在内部，压缩表现为时间轴上的特殊提交。

ROLLBACK：指示提交/增量提交失败并回滚，删除在此写入过程中产生的所有部分文件。

SAVEPOINT：将某些文件组标记为“已保存”，以便清理器不会删除它们。在发生灾难/数据恢复场景时，它有助于将表恢复到时间轴上的某个点。

Instant time（即时时间）

即时时间通常是一个时间戳(例如:20190117010349)，它按动作开始时间的顺序单调增加。有两个重要时间概念

Arrival time：数据到达Hudi的时间。

Event Time：数据记录中的时间。

State：瞬时的当前状态。

REQUESTED：表示一个action已经调度，但尚未执行。

INFLIGHT：表示当前action正在执行。

COMPLETED：表示时间轴上action已完成。

冲突检查会在 instant 状态变换的两个节点进行，一个是 requested 转 inflight 状态，一个是 inflight 转 completed 状态。其中，后者状态变换时，会进行加锁操作，以实现版本隔离。
冲突检查即是对 instant 创建到状态变化的过程中其他已经完成/正在执行的 instant 之间的进行冲突检查，检查策略分为行列两种。

行级别的冲突检查即是不能同时有两个 instant 往同一个 file group 写。
列级别的冲突检查即是可以有两个 instant 往同一个 file group 写，但是两个 instant 写入的schema 不可以存在交集。
每个 instant 只写入 schema 中的部分列，log 文件中的数据只包含 schema 中的部分
Compaction 按主键拼接不同列下的数据，Parquet 文件中存储的数据拥有完整的 schema

2.2、upsert支持

hudi支持对数据的upsert操作，对于upsert操作的支持是通过hudi的文件组织特性保证的。hudi中分为COW和MOR两种类型的表，upsert操作的时候，会根据每条数据的record key进行定位。

COW表：使用专门的列式文件格式存储数据，更新时保存多版本，并且在写的过程中通过异步的Merge来实现重写（Rewrite）数据文件。
COW表只包含列式格式的Base文件，每次执行COMMIT操作会生成新版本的Base文件，最终执行COMPACTION操作时还是会生成列式格式的Base文件。每次执行INSERT或UPDATE操作，都会在Timeline上生成一个的COMMIT，同时对应着一个文件分片（File Slice）。如果是INSERT操作则生成文件分组的第一个新的文件分片，如果是UPDATE操作则会生成一个新版本的文件分片。

MOR表：使用列式和行式文件格式混合的方式来存储数据，列式文件格式比如Parquet，行式文件格式比如Avro。更新时写入到增量（Delta）文件中，之后通过同步或异步的COMPACTION操作，生成新版本的列式格式文件。
Merge-On-Read表存在列式格式的Base文件，也存在行式格式的增量（Delta）文件，新到达的更新都会写到增量日志文件中，根据实际情况进行COMPACTION操作来将增量文件合并到Base文件上。通常，需要有效的控制增量日志文件的大小，来平衡读放大和写放大的影响。
下图中，每个文件分组都对应一个增量日志文件（Delta Log File）。COMPACTION操作在后台定时执行，会把对应的增量日志文件合并到文件分组的Base文件中，生成新版本的Base文件。

3、Hudi 表存储类型及比较

3.1、前序

Hudi提供两类型表：写时复制(Copy on Write, COW)表和读时合并(Merge On Read, MOR)表。

对于Copy-On-Write Table，用户的update会重写数据所在的文件，所以是一个写放大很高，但是读放大为0，适合写少读多的场景。

对于Merge-On-Read Table，整体的结构有点像LSM-Tree，用户的写入先写入到delta data中，这部分数据使用行存，这部分delta data可以手动merge到存量文件中，整理为parquet的列存结构。

3.2、数据计算模型

hudi是Uber主导开发的开源数据湖框架，所以大部分的出发点都来源于Uber自身场景，比如司机数据和乘客数据通过订单ID来做join等。

在hudi过去的使用场景里，和大部分公司的架构类似，采用批式和流式共存的Lambda架构，后来Uber提出增量Incremental模型，相对批式来讲，更加实时，相对流式而言，更加经济。

3.2.1、批式模型(Batch)

批式模型就是使用MapReduce、Hive、Spark等典型的批计算引擎，以小时任务或者天任务的形式来做数据计算

特性

A.延迟：小时级延迟或者天级别延迟。这里的延迟不单单指的是定时任务的时间，在数据架构里，这里的延迟时间通常是定时任务间隔时间+一系列依赖任务的计算时间+数据平台最终可以展示结果的时间。数据量大、逻辑复杂的情况下，小时任务计算的数据通常真正延迟的时间是2-3小时。

B.数据完整度：数据较完整。以处理时间为例，小时级别的任务，通常计算的原始数据已经包含了小时内的所有数据，所以得到的数据相对较完整。但如果业务需求是事件时间，这里涉及到终端的一些延迟上报机制，在这里，批式计算任务就很难派上用场。

C.成本：成本很低。只有在做任务计算时，才会占用资源，如果不做任务计算，可以将这部分批式计算资源出让给在线业务使用。从另一个角度来说成本是挺高的，如原始数据做了一些增删改查，数据晚到的情况，那么批式任务是要全量重新计算。

3.2.2、流式模型(Stream)

流式模型，典型的就是使用Flink来进行实时的数据计算

特性

A.延迟：很短，甚至是实时。

B.数据完整度：较差。因为流式引擎不会等到所有数据到齐之后再开始计算，所以有一个watermark的概念，当数据的时间小于watermark时，就会被丢弃，这样是无法对数据完整度有一个绝对的保障。在互联网场景中，流式模型主要用于活动时的数据大盘展示，对数据的完整度要求并不算很高。在大部分场景中，用户需要开发两个程序，一是流式数据生产流式结果，而是批式计算人物，用于次日修复实时结果

C.成本：很高。因为流式任务时常驻的，并且对于多流join的场景，通常要借助内存或者数据库来做state的存储，不管是序列化开销，还是和外部组件交互产生的额外IO，在大数据量下都是不容忽视的。

3.2.3、增量模型

简单来讲，就是一mini batch的形式来跑准实时任务。hudi在增量模型中支持了两个最重要的特性：

A.Upsert：这个主要是解决批式模型中，数据不能插入、更新的问题，有了这个特性，可以往Hive中写入增量数据，而不是每次进行完全的覆盖。（hudi自身维护了key-file的映射，所以当upsert时很容易找到key对应的文件）

B.Incremental Query：增量查询，减少计算的原始数据量。以uber中司机和乘客的数据流join为例，每次抓取两条数据流中的增量数据进行批式的join即可，相比流式数据而言，成本要降低几个数量级。

3.3、查询类型（Query Type）

Hudi支持三种不同的查询表的方式：Snapshot Queries(快照查询)、Incremental Queries(增量查询)和Read Optimized Queries(读优化查询).

3.3.1、Snapshot Queries(快照查询)

查询某个增量提交操作中数据集的最新快照，先进行动态合并最新的基本文件（parquet）和增量文件（Avro）来提供近实时数据集（通常会存在几分钟的延迟）

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

3.3.2、Incremental Queries(增量查询)

仅查询新写入数据集的文件，需要指定一个Commit/Compaction的即时时间（位于Timeline上的某个instant）作为条件，来查询此条件之后的新数据

可查看自给定commit/delta commit即时操作依赖新写入的数据，有效地提供变更流来启用增量数据管道

3.3.3、Read Optimized Queries(读优化查询)

直接查询基本文件（数据集的最新快照），其实就是列式文件（Parquet）。并保证与非hudi列式数据集相比，具有相同的列式查询性能

可查看给定的commit/compact即时操作的表的最新快照

读优化查询和快照查询相同仅访问基本文件，提供给定文件片自上次执行压缩操作以来的数据。通常查询数据的最新程度的保证取决于压缩策略

3.4、表类型解析

3.4.1、Copy On Write

简称COW，它实在数据写入的时候，复制一份原来的拷贝，在其基础上添加新数据，正在读数据的请求，读取的是最近的完整副本，这类似于MySQL的MVCC思想

优点：读取时，只读取对应分区的一个数据文件即可，较为高效

*缺点：数据写入的时候，需要复制一个先前的副本再在其基础上生成新的数据文件，这个过程比较耗时

COW表主要使用列式文件格式（parquet）存储数据，在写入数据过程中，执行同步合并，更新数据版本并重写数据文件，类似RDBMS中的B-Tree更新

A.更新update：在更新记录时，hudi会先找到包含更新数据的文件，然后再使用更新值（最新的数据）重写该文件，包含其他记录的文件保持不变。当突然有大量写操作时会导致重写大量文件，从而导致极大的IO开销

B.读取read：在读取数据时，通过读取最新的数据文件来获取最新的更新，此存储类型适用于少量写入和大量读取的场景

3.4.2、Merge On Read

简称MOR，新插入的数据存储在delta log中，定期再将delta log合并进行parquet数据文件，读取数据时，会将delta log跟老的数据文件做merge，得到完整的数据返回

*优点：由于写入数据先写delta log，且delta log较小，所以写入成本较低

*缺点：需要定期合并整理compact，否则碎片文件较多。读取性能较差，因为需要将delta log和老数据文件合并

MOR表是COW表的升级版，他使用列式（parquet）与行式（avro）文件混合的方式存储数据。在更新记录时，类似nosql中的LSM-Tree更新

A.更新：在更新记录时，仅更新到增量文件（avro）中，然后进行异步（或同步）合并到新文件，最后创建列式文件（parquet）的新版本。次存储类型适合频繁写的工作负载，因为新纪录是以追加的模式写入增量文件中

B.读取：在读取数据集时，需要先将增量文件与旧文件进行合并，然后圣承列式文件成功后，在进行查询

3.5、COW vs MOR

对于写时复制（COW）和读时合并（MOR）writer来说，Hudi的WriteClient是相同的。COW表：用户在snapshot读取的时候会扫描所有最新的FileSlice的base file，MOR表：在READ OPTIMIZED模式下，只会读最近的经过compaction的commit

4、timeline 增量查询

为了支持增量查询，Hudi使用时间轴（Timeline）功能来跟踪表的变更历史，并记录每个操作的增量包含的文件路径和时间戳信息。通过时间轴和查询引擎，Hudi可以组合不同的数据文件，查找给定时间戳或时间范围内的所有增量操作。这样，即使在COW表中没有Delta文件，Hudi仍然可以跟踪和查询表的更新历史记录。
COW表的增量查询：

Hudi引擎会首先查询时间轴（Timeline），以查找给定时间戳或时间范围内涉及到的所有文件路径（File Paths）
读取每个数据文件时，Hudi引擎需要使用该文件的元数据信息（File Metadata）来定位增量数据记录和相关的记录键（Record Key）。parquet的File Metadata记录了该文件中包含的所有记录的元数据信息，包括**记录键（Record Key）、时间戳（Timestamp）**等相关信息。增量查询时引擎会先解析数据文件元数据，并使用其中的时间戳和记录键信息来过滤出符合条件的记录。
当Hudi引擎从数据文件中提取出指定时间戳或时间范围内的增量数据记录之后，剩下的查询工作就是标准的数据记录查询。Hudi引擎会使用记录键（Record Key）和时间戳（Timestamp）的组合来唯一标识每个记录和其对应的增量数据历史记录，并在查询时使用这些信息来定位和选择特定的增量数据记录，最终返回给用户符合条件的查询结果。

5、事务

Hudi的事务功能被称为Timeline，因为Hudi把所有对一张表的操作都保存在一个时间线对象里面。Hudi官方文档中对于Timeline功能的介绍稍微有点复杂，不是很清晰。其实从用户角度来看的话，Hudi提供的事务相关能力主要是这些：

hudi基于timeline实现事务，timeline上每个instant会包含当前版本对于的file path list。当client读取数据时，首先会查看timeline里最新的commit是哪个，从最新的commit里获得对应的文件列表，再去这些文件读取真正的数据。

Hudi通过这种方式实现了多版本隔离的能力。当一个client正在读取v1的数据时，另一个client可以同时写入新的数据，新的数据会被写入新的文件里，不影响v1用到的数据文件。只有当数据全部写完以后，v2才会被commit到timeline里面。后续的client再读取时，读到的就是v2的数据。

顺带一提的是，尽管Hudi具备多版本数据管理的能力，但旧版本的数据不会无限制地保留下去。Hudi会在新的commit完成时开始清理旧的数据，默认的策略是“清理早于10个commit前的数据”。
hudi事务和增量查询原理

Hudi主要设计
Hudi相较与传统数仓的TableStructre主要做了以下设计

定义RecordKey，每个Record必须有唯一的主键
分区的文件划分成多个FileGroup(表观上是具有相同前缀的一组文件)，同一个RecordKey只能属于一个FileGroup
FileGroup内的文件分为BaseFile和DeltaFile。DeltaFile记录对BaseFile的修改。多版本的BaseFile可能会同时存在；Compaction任务会把DeltaFile和BaseFile合并成新的BaseFile
引入.hoodie元数据文件夹，主要记录commit文件，每次数据导入/Compaction等对文件结构的修改都必须提交commit文件来确保持久化

6、Hudi主要设计

Hudi相较与传统数仓的TableStructre主要做了以下设计

定义RecordKey，每个Record必须有唯一的主键
分区的文件划分成多个FileGroup(表观上是具有相同前缀的一组文件)，同一个RecordKey只能属于一个FileGroup
FileGroup内的文件分为BaseFile和DeltaFile。DeltaFile记录对BaseFile的修改。多版本的BaseFile可能会同时存在；Compaction任务会把DeltaFile和BaseFile合并成新的BaseFile
引入.hoodie元数据文件夹，主要记录commit文件，每次数据导入/Compaction等对文件结构的修改都必须提交commit文件来确保持久化

8.1、文件版本

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

8.2、文件格式

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的包含数据文件头（File Footer）和数据文件元数据（File Metadata）
在 footer 的 meta 中记录了 record key 组成的 BloomFilter，用于在 file based index 的实现中实现高效率的 key contains 检测。只有不在 BloomFilter 的 key 才需要扫描整个文件消灭假阳。
在File Metadata中记录了文件中包含的所有记录的元数据信息，包括每个记录的记录键（Record Key）和时间戳（Timestamp），以及其他与记录相关的信息。通过读取和解析数据文件元数据，Hudi可以确定每个数据文件中包含哪些增量写记录，以及每个记录的更新时间戳和其他相关的元数据信息。

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

8.3、Index

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

Hoodie key (record key + partition path) 和 file id (FileGroup) 之间的映射关系，数据第一次写入文件后保持不变，所以，一个 FileGroup 包含了一批 record 的所有版本记录。Index 用于区分消息是 INSERT还是 UPDATE。
Index创建过程：

新增 records 找到映射关系：record key => target partition
当前最新的数据 找到映射关系：partition => (fileID, minRecordKey, maxRecordKey) LIST （如果是 base files 可加速）
新增 records 找到需要搜索的映射关系：fileID => HoodieKey(record key + partition path) LIST，key 是候选的 fileID
通过 HoodieKeyLookupHandle 查找目标文件（通过 BloomFilter 加速）

索引类型

Bloom Index：通过BloomFilter快速定位可能的FileGroup，减少数据扫描的范围，大数据量可能存在假阳问题。
Hbase Index：将HoodieKey与FileGroup的关系放到Hbase中，在插入FileGroup定位阶段所有task向Hbase发送Batch Get请求，获取Record Key的mapping信息。对小批次的keys查询效率高，但是需要引入外部系统，增加运维压力。
Simple Index：将需要更新的key与base文件的key进行join，性能较差
Hive Bucket Index：为避免Bloom Index的假阳引入的分桶索引，将Record Key打散为多个Bucket，每个Bucket对应一个FileGroup。理论上有Bucket会出现数据倾斜。
Non Index：字节针对日志数据的优化，适合无主键的日志数据导入，数据不会根据主键进行去重，支持高效的Append操作。
Flinke State：HUDI 在 0.8.0 版本中实现 的 Flink witer，采用了 Flink的 state 作为底层的 index 存储，每个 records 在写入之前都会先计算目标 bucket ID。Flink是基于状态计算，如果索引数据特别大，进一步影响Flink的CK，另一部分会影响Flink资源的使用，可以进行状态调优
注意：Flink 只有一种 state based index（和 bucket_index），其他 index 是 Spark 可选配置。
Bloom Index的具体流程：

将要插入的Record以RecordKey为标示构建RDD[RecordKey]，大小为M
分区内全部BaseFile集合构建RDD[BaseFile]，大小为N
两个集合做笛卡尔积，生成一个RDD[RecordKey,BaseFile]，大小为M*N。
每个(RecordKey,BaseFile),加载BaseFile对应的BloomFilter，判断RecordKey是否存在
如果存在，加载BaseFile的所有Key判断是否精确命中
org.apache.hudi.index.bloom.SparkHoodieBloomIndex

8.4、文件布局一

8.4.1、说明

探索Apache Hudi核心概念 (1) - File Layouts一文。该文章对Hudi File layout的讲解非常详细

Hudi官网https://hudi.apache.org/docs/file_layouts/对Hudi文件布局的描述如下：

1）Hudi表的所有内容（数据文件和元数据）在分布式文件系统下一个指定的目录中保存（称之为bash path）。

2）Hudi表中的数据按照分区来存储（体现为以分区字段值命名的目录）。

3）在每个分区中数据文件按照file group来组织。属于同一个file group的数据文件具有相同的file ID。

4）每个file group包含多个file slice。

5）每个file slice包含一个base file和一系列log file（仅MOR表有log file）。Base file的存储格式由hoodie.table.base.file.format确定，由commit或compaction操作生成。Log file是对应的base file的增量数据，由deltacommit操作生成。

8.4.2、基本原则

“Hudi在文件操作上有一个重要“原则”：Hudi always creates immutable files on disk。即：文件一旦创建，永远不会再更新，任何添加、修改或删除操作只会在现有文件数据的基础上合并输入数据一起写入到下一个新文件中。”

Hudi的数据文件位于表base path中。如果有分区，数据文件位于base path的分区目录中。在同一个分区目录中，Hudi的数据文件没有层级划分。仅在逻辑上分为file slice和file group。

Hudi上存储中所有的文件都是一次性写入的。对于Base file（列存储），任何修改会创建出新的file slice，和原来的file slice的file id相同，属于同一个file group。但是它比原来的file slice时间戳更大，相当于“替换”了原来的file slice内容。保留原file slice的原因是可以通过time travel query等回溯历史数据。

8.4.3、Base file

“Base File是存储Hudi数据集的主体文件，以Parquet等列式格式存储，所以我们在Hudi中看到的Parquet文件基本都是Base File。实际上，Base File的命名是为了呼应Log File，在没有Log File的COW表里，Base File就是基层的数据存储文件，没必要强调它的“Base”身份，直接叫Parquet文件就可以。Base File遵循一致的命名规范，格式为：

<fileId>_<writeToken>_<instantTime>.parquet

fileId部分是一个uuid，我们会在多个文件中看到相同的fileId，这些fileId相同的文件就组成了一个File Group。instantTime是写入这个文件对应的instant的时间，也是该文件的一个“版本”标注，因为一个Base File历经多轮增删改操作后就会产生多个版本，Hudi就使用instantTime对它们进行标识。不管是MOR表还是COW表，都有Base File，只是在COW表里只有Base File，在MOR表里除了Base File还有Log File。”

8.4.4、Log file

“Log File是在MOR表中用于存储变化数据的文件，也常被称作Delta Log，Log File不会独立存在，一定会从属于某个Parquet格式的Base File，一个Base File和它从属的若干Log File所构成的就是一个File Slice。Log File也遵循一致的命名规范，格式为：

.<fileId>_<baseCommitTime>.log.<fileVersion>_<writeToken>

不同于Base File，Log File文件名中时间戳部分并不是Log File自己对应的instanceTime，而是它所从属的Base File的instanceTime，即baseCommitTime。如此一来，就没有办法通过时间戳来区分Log File提交的先后顺序了，所以Hudi在Log File文件名中加入了fileVersion，它是一个从1开始单调递增的序列号，用于标识Log File产生的顺序。”

Log file可以理解为base file的增量部分。在base file更新之前起到了数据缓存的作用。

8.4.5、File Slice

“在MOR表里，由一个Base File和若干从属于它的Log File组成的文件集合被称为一个File Slice。应该说File Slice是针对MOR表的特定概念，对于COW表来说，由于它不生成Log File，所以File Silce只包含Base File，或者说每一个Base File就是一个独立的File Silce。总之，对于COW表来说没有必要区分File Silce，也不没必要强调Base File的“Base”身份，只是为了概念对齐，大家会统一约定Hudi文件的三层逻辑布局为：File Group -> File Slice -> Base / Log Files。”

File slice对应了一批数据的某一个版本。file id相同的多个file slice分别对应同一批数据的不同版本。使用时间戳来区分数据的新旧。不同file id的file slice中不会有相同数据的多个版本。

8.4.6、File Group

“在前面介绍Base File时，我们已经提到了File Group，简单说，就是fileId相同的文件属于同一个File Group。同一File Group下往往有多个不同版本（instantTime）的Base File（针对COW表）或Base File + Log File的组合（针对MOR表），当File Group内最新的Base File迭代到足够大（ >100MB）时，Hudi就不会在当前File Group上继续追加数据了，而是去创建新的File Group。”

同一个File Group中保存的是同一批数据的不同版本。在File Group中的旧数据文件（时间戳不为最新的base file）保存了这批数据的历史版本。

不同的File Group保存了不同的数据。因此查询最新数据的时候，同一个File Group中的文件只需要考虑最新版本的file slice。将不同file group的查询结果合并返回。

Hudi有文件大小控制功能。会尽力保持base file大小在配置的范围内。因此如果file group过小，新数据会优先追加到较小的file group中，创建新file group（数据切分）的逻辑是单个file group的数据量超过了配置上限。

每个file group中的数据分布可以通过clustering表服务来优化。将数据按照经常作为查询条件的字段，依照指定顺序排序存放。

8.4.7、Hudi表目录示意

MOR表：

-rw-r--r--   3 hdfs hdfs       4283 2024-07-09 13:46 /hudi_student/.ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_20240709133903537.log.1_0-1-0
-rw-r--r--   3 hdfs hdfs       4871 2024-07-10 08:42 /hudi_student/.ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_20240709134613722.log.1_0-1-0
-rw-r--r--   3 hdfs hdfs       4074 2024-07-10 08:44 /hudi_student/.ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_20240710084244018.log.1_0-1-0
-rw-r--r--   3 hdfs hdfs        815 2024-07-10 14:20 /hudi_student/.ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_20240710084455963.log.1_0-1-0
drwxr-xr-x   - hdfs hdfs          0 2024-07-10 14:20 /hudi_student/.hoodie
-rw-r--r--   3 hdfs hdfs         96 2024-07-09 13:39 /hudi_student/.hoodie_partition_metadata
-rw-r--r--   3 hdfs hdfs     434687 2024-07-09 17:14 /hudi_student/ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_0-1-0_20240709134613722.parquet
-rw-r--r--   3 hdfs hdfs     434725 2024-07-10 08:42 /hudi_student/ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_0-1-0_20240710084244018.parquet
-rw-r--r--   3 hdfs hdfs     434756 2024-07-10 08:44 /hudi_student/ba74ba57-d45c-43c7-9ddb-7c8afb3bab8f_0-1-0_20240710084455963.parquet

COW表：

drwxr-xr-x   - hdfs hdfs          0 2024-07-09 18:13 /hudi_student_cow/.hoodie
-rw-r--r--   3 hdfs hdfs         96 2024-07-09 17:25 /hudi_student_cow/.hoodie_partition_metadata
-rw-r--r--   3 hdfs hdfs     434759 2024-07-09 17:25 /hudi_student_cow/c0094994-a251-4933-8de6-6858da510517_0-1-0_20240709172551274.parquet
-rw-r--r--   3 hdfs hdfs     434752 2024-07-09 17:55 /hudi_student_cow/c0094994-a251-4933-8de6-6858da510517_0-1-0_20240709175507082.parquet
-rw-r--r--   3 hdfs hdfs     434758 2024-07-09 18:04 /hudi_student_cow/c0094994-a251-4933-8de6-6858da510517_0-1-0_20240709180410695.parquet
-rw-r--r--   3 hdfs hdfs     434757 2024-07-09 18:13 /hudi_student_cow/c0094994-a251-4933-8de6-6858da510517_0-1-0_20240709181300182.parquet

8.5、文件布局详解二

Hudi在分布式文件系统的基础路径下将数据表组织成目录结构

表被按照分区进行切分，每个分区在基础路径下为一个子目录
在每个分区内，文件被组织成File Group，由File ID唯一标识
每个File Group包含多个File Silce
每个File Silce都包含一个在某个Instance生成的BaseFile(.parquet)，以及一组LogFile(.log.*)

.hoodie 目录是 Hudi 表的核心目录，它包含了 Hudi 表的元数据和其他相关文件和目录
.temp 目录用于存储正在写入的数据
.tmp 目录用于存储已完成写入但尚未提交的数据
archive 目录用于存储归档数据
metadata 目录包含了所有分区的元数据信息
timeline.json 文件包含了表的时间轴信息
version 文件包含了表的版本信息
write.lock 文件用于控制并发写入
.hoodie_partition_metadata 文件包含了该分区的元数据信息，例如分区键、分区路径等。
分区目录是按照分区键组织的目录，每个分区目录下都包含了该分区下的所有数据文件和 .hoodie_partition_metadata 文件。

8.6、Hudi写入流程

COW

先对 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

2、MOR

先对 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 一个新的

COW和MOR对比
COW

场景：适合离线场景(写少读多)
存储：列存(Parquet)
写入：每次写入数据，会先读取已有数据文件，然后与更新数据合并写入新的文件
查询：支持Hive表类似的HQL查询
MOR

场景：更适合近实时/实时场景(写多读少)
存储：行存(Delta Log)+列存(Base File)
写入：每次数据写入会先Append进行行存(Delta Log)，写入指定次数后，行存文件会与列存文件(BaseFile)进行合并生产新的列存文件，即Compaction过程
查询：
RealTime查询：读取行存+列存，数据延迟低
ReadOptimized查询：近读取列存数据，查询性能高
Incremental查询：适合用于增量消费场景(需要指定时间戳)

Hudi Compact操作
没有 base file：走 copy on write insert 流程，直接 merge 所有的 log file 并写 base file
有 base file：走 copy on write upsert 流程，先读 log file 建 index，再读 base file，最后读 log file 写新的 base file
Flink 和 Spark streaming 的 writer 都可以 apply 异步的 compaction 策略，按照间隔 commits 数或者时间来触发 compaction 任务，在独立的 pipeline 中执行。

总结
通过对写流程的梳理我们了解到 HUDI 相对于其他数据湖方案的核心优势：

写入过程充分优化了文件存储的小文件问题，Copy On Write 写会一直将一个 bucket （FileGroup）的 base 文件写到设定的阈值大小才会划分新的 bucket；Merge On Read 写在同一个 bucket 中，log file 也是一直 append 直到大小超过设定的阈值 roll over。
对 UPDATE 和 DELETE 的支持非常高效，一条 record 的整个生命周期操作都发生在同一个 bucket，不仅减少小文件数量，也提升了数据读取的效率（不必要的 join 和 merge）。
hudi增删改查基础api

import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions}
import org.apache.hudi.common.table.HoodieTableConfig
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.functions.{col, concat_ws}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import scala.collection.mutable

object HudiApiTest {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().enableHiveSupport().getOrCreate()

    /**
     * 写hudi
     * BULK_INSERT_OVERWRITE_OPERATION_OPT_VAL指定分区overwrite语意，
     * 或者通过DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "date,hour"指定分区字段
     * mode必须是SaveMode.Append，SaveMode.Overwrite会覆盖整个表
     */
    val insertDF = spark.sql(
      """
        |select md5(concat('','.','dp_compliance','.','hoodie_test')) as id,'' as cluster_name,'dp_compliance' as database_name,'hoodie_test' as table_name,'20230518' as date,'00' as hour
        |union all
        |select md5(concat('','.','dp_compliance','.','hoodie_test2')) as id,'' as cluster_name,'dp_compliance' as database_name,'hoodie_test2' as table_name,'20230518' as date,'00' as hour
        |""".stripMargin)
    val configs = new mutable.HashMap[String, String]()
    configs += (HoodieTableConfig.HOODIE_TABLE_NAME_PROP_NAME -> "test_bytelake")
    configs += (HoodieTableConfig.HOODIE_DATABASE_NAME_PROP_NAME -> "dp_compliance_test")
//    指定分区字段
//    configs += (DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "date,hour")
//    或者动态分区
    configs += (DataSourceWriteOptions.OPERATION_OPT_KEY -> DataSourceWriteOptions.BULK_INSERT_OVERWRITE_OPERATION_OPT_VAL)

    insertDF.write
    .format("hudi")
    .options(configs)
    .mode(SaveMode.Append)  //BULK_INSERT_OVERWRITE_OPERATION_OPT_VAL控制分区overwrite语意
    .save()

    /**
     * 查询数据
     */
    spark
      .read
      .format("hudi")
      .load("hdfs://harunava/home/byte_dw_compliance/warehouse/dp_compliance_test.db/test_bytelake/*/*")
      .show()

    /**
     * 更新数据
     * 更新数据和insert相同，hudi根据主键recordkey更新数据，保留最新的一条
     */
    val updateDF = spark
      .sql(
        """
          |select md5(concat('','.','dp_compliance','.','hoodie_test2')) as id,'' as cluster_name,'dp_compliance' as database_name,'hoodie_test3' as table_name,'20230518' as date,'00' as hour
          |""".stripMargin)
    updateDF.write
      .format("hudi")
      .options(configs)
      .mode(SaveMode.Append)  //BULK_INSERT_OVERWRITE_OPERATION_OPT_VAL控制分区overwrite语意
      .save()

    /**
     * 增量查询
     * 指定数据查询方式，有以下三种：
     * val QUERY_TYPE_SNAPSHOT_OPT_VAL = "snapshot"    -- 获取最新所有数据 , 默认
     * val QUERY_TYPE_INCREMENTAL_OPT_VAL = "incremental"  --获取指定时间戳后的变化数据
     * val QUERY_TYPE_READ_OPTIMIZED_OPT_VAL = "read_optimized"  -- 只查询Base文件中的数据
     */
    spark
      .read
      .format("hudi")
      .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY,DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
//      指定查询某个时间戳之前提交的数据，依据_hoodie_commit_time筛选
      .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY,"20230520071825")
      .load("/home/byte_dw_compliance/warehouse/dp_compliance_test.db/test_bytelake/*/*/")
      .show()

    spark
      .read.format("hudi")
      .option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY,DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
      //指定查询开始时间（不包含），“000”指定为最早时间
      .option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, "00000")
      //指定查询结束时间（包含）
      .option(DataSourceReadOptions.END_INSTANTTIME_OPT_KEY, "20230520071825")
      .load("/home/byte_dw_compliance/warehouse/dp_compliance_test.db/test_bytelake/*/*/")
      .show()

    /**
     * 删除数据
     * 删除时根据分区和主键定位，都相同时删除数据
     */
    val deleteDF = spark.sql(
      """
        |select md5(concat('','.','dp_compliance','.','hoodie_test')) as id,'20230518' as date,'00' as hour
        |union all
        |select md5(concat('','.','dp_compliance','.','hoodie_test2')) as id,'20230518' as date,'01' as hour
        |""".stripMargin)

    deleteDF.write.format("hudi")
      //指定表名，这里的表明需要与之前指定的表名保持一致
      .option(HoodieTableConfig.HOODIE_DATABASE_NAME_PROP_NAME,"dp_compliance_test")
      .option(HoodieTableConfig.HOODIE_TABLE_NAME_PROP_NAME,"test_bytelake")
      //指定操作模式为delete
      .option(DataSourceWriteOptions.OPERATION_OPT_KEY,DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL)
      //指定分区字段
      .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY,"date,hour")
      //设置删除并行度设置，默认1500并行度
      .option("hoodie.delete.shuffle.parallelism", "2")
      .mode(SaveMode.Append).save()
  }
}