1.1、介绍
Overview | Apache Hudi!Welcome to Apache Hudi! This overview will provide a high level summary of what Apache Hudi is and will orient you onhttps://hudi.apache.org/docs/overview

Apache Hudi（Hadoop Upserts Delete and Incremental）是下一代流数据湖平台。Apache Hudi将核心仓库和数据库功能直接引入数据湖。Hudi提供了表、事务、高效的upserts/delete、高级索引、流摄取服务、数据集群/压缩优化和并发，同时保持数据的开源文件格式。

Apache Hudi不仅非常适合于流工作负载，而且还允许创建高效的增量批处理管道。

Apache Hudi可以轻松地在任何云存储平台上使用。Hudi的高级性能优化，使分析工作负载更快的任何流行的查询引擎，包括Apache Spark、Flink、Presto、Trino、Hive等。


1.2、发展历史

2015 年：发表了增量处理的核心思想/原则（O'reilly 文章）。
2016 年：由 Uber 创建并为所有数据库/关键业务提供支持。
2017 年：由 Uber 开源，并支撑 100PB 数据湖。
2018 年：吸引大量使用者，并因云计算普及。
2019 年：成为 ASF 孵化项目，并增加更多平台组件。
2020 年：毕业成为 Apache 顶级项目，社区、下载量、采用率增长超过 10 倍。
2021 年：支持 Uber 500PB 数据湖，SQL DML、Flink 集成、索引、元服务器、缓存。
1.3、特性

可插拔索引机制支持快速Upsert/Delete。
支持增量拉取表变更以进行处理。
支持事务提交及回滚，并发控制。
支持Spark、Presto、Trino、Hive、Flink等引擎的SQL读写。
自动管理小文件，数据聚簇，压缩，清理。
流式摄入，内置CDC源和工具。
内置可扩展存储访问的元数据跟踪。
向后兼容的方式实现表结构变更的支持。

1.4、使用场景

近实时写入
减少碎片化工具的使用。
CDC 增量导入 RDBMS 数据。
限制小文件的大小和数量。
近实时分析
相对于秒级存储（Druid, OpenTSDB），节省资源。
提供分钟级别时效性，支撑更高效的查询。
Hudi作为lib，非常轻量。
增量 pipeline
区分arrivetime和event time处理延迟数据。
更短的调度interval减少端到端延迟（小时 -> 分钟） => Incremental Processing。
增量导出
替代部分Kafka的场景，数据导出到在线服务存储 e.g. ES。

2、概要
2.1、时间轴（Timeline）


hudi的核心是维护在不同时刻在表上执行的所有操作的时间表，提供表的即时视图，同时还有效地支持按时间顺序检索数据。Hudi的时刻由以下组件组成：

Instant action: 在表上执行的操作类型
commits: 表示将一批数据原子写入表中
cleans: 清除表中不在需要的旧版本文件的后台活动。
delta_commit:增量提交是指将一批数据原子性写入MergeOnRead类型的表中，其中部分或者所有数据可以写入增量日志中。
compaction: 合并Hudi内部差异数据结构的后台活动，例如:将更新操作从基于行的log日志文件合并到列式存储的数据文件。在内部，COMPACTION体现为timeline上的特殊提交。
rollback:表示提交操作不成功且已经回滚，会删除在写入过程中产生的数据
savepoint:将某些文件标记为“已保存”，以便清理程序时不会被清楚。在需要数据恢复的情况下，有助于将数据集还原到时间轴上某个点。
Instant time:即时时间
通常是一个时间戳（例如：20230422210349），它按照动作开始时间的顺序单调增加。
State: 时刻的当前状态
requested:表示一个动作已被安排，但尚未启。
inflight:表是当前正在执行操作。
completed:表是在时间线上完成了操作。

两个时间概念

区分两个重要的时间概念：

Arrival time: 数据到达 Hudi 的时间，commit time。

Event time: record 中记录的时间。


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

Hudi将一个表映射为如下文件结构


Hudi存储分为两个部分：

1、元数据：.hoodie目录对应着表的元数据信息，包括表的版本管理（Timeline）、归档目录（存放过时的instant也就是版本），一个instant记录了一次提交（commit）的行为、时间戳和状态，Hudi以时间轴的形式维护了在数据集上执行的所有操作的元数据；
2、数据：和hive一样，以分区方式存放数据；分区里面存放着Base File（.parquet）和Log File（.log.*）；


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

（2）表被划分为多个分区，这些分区是包含该分区的数据文件的文件夹，非常类似于Hive表

（3）在每个分区中，文件被组织成文件组，由文件ID唯一标识

（4）每个文件组包含几个文件片（FileSlice）

（5）每个文件片包含：

一个基本文件(.parquet)：在某个commit/compaction即时时间（instant time）生成的（MOR可能没有

多个日志文件(.log.*)，这些日志文件包含自生成基本文件以来对基本文件的插入/更新（COW没有）

（6）Hudi采用了多版本并发控制(Multiversion Concurrency Control, MVCC)

compaction操作：合并日志和基本文件以产生新的文件

clean操作：清除不使用的/旧的文件片以回收文件系统上的空间




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

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

MVCC（Multi-Version Concurrency Control）：多版本并行发控制机制

Multi-Versioning:产生多版本的数据内容，使得读写可以不互相阻塞

Concurrency Control：并发控制，使得并行执行的内容能保持串行化结果
2.3、索引（Index）
2.3.1、原理


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

Hudi为了消除不必要的读写，引入了索引的实现。在有了索引之后，更新的数据可以快速被定位到对应的FileGroup。上图为例，白色是基本文件，黄色是更新数据，有了索引机制，可以做到：避免读取不需要的文件、避免更新不必要的文件、无需将更新数据与历史数据做分布式关联，只需要在FileGroup内做合并。
2.3.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可选配置。
2.3.3、全局索引与非全局索引

全局索引：全局索引在全表的所有分区范围下强制要求键的唯一性，也就是确保对给定的键有且只有一个对应的记录。全局索引提供了更强的保证，但是随着表增大，update/delete操作损失的性能越高，因此更适用于小表。

非全局索引：默认的索引实现，只能保证数据在分区的唯一性。非全局索引依靠写入器为同一个记录的update/delete提供一致的分区路径，同时大幅提高了效率，更适用于大表。
从index的维护成本和写入性能的角度考虑，维护一个globalindex的难度更大，对写入性能的影响也更大，所以需要non-globalindex。

HBase索引本质上是一个全局索引，bloom和simpleindex都有全局选项：

hoodie.index.type=GLOBAL_BLOOM

hoodie.index.type=GLOBAL_SIMPLE

2.3.4、索引的选择策略
1、对事实表的延迟更新

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




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

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

2、对事件表的去重

事件流无处不在。从ApacheKafka或其他类似的消息总线发出的事件数通常是事实表大小的10-100倍。事件通常把时间（到达时间、处理时间）作为首类处理对象，比如物联网的事件流、点击流数据、广告曝光数等等。由于这些大部分都是仅追加的数据，插入和更新只存在于最新的几个分区中。由于重复事件可能发生在整个数据管道的任一节点，在存放到数据湖前去重是一个常见的需求。




总的来说，低消耗去重是一个非常有挑战的工作。虽然可以用一个键值存储来实现去重（即HBase索引），但索引存储的消耗会随着事件数增长而线性增长以至于变得不可行。事实上，有范围裁剪功能的布隆索引是最佳的解决方案。我们可以利用作为首类处理对象的时间来构造由事件时间戳和事件id（event_ts+event_id)组成的键，这样插入的记录就有了单调增长的键。这会在最新的几个分区里大幅提高裁剪文件的效益。

3、对维度表的随机更删




正如之前提到的，如果范围比较不能裁剪许多文件的话，那么布隆索引并不能带来很好的效益。在这样一个随机写入的作业场景下，更新操作通常会触及表里大多数文件从而导致布隆过滤器依据输入的更新对所有文件标明阳性。最终会导致，即使采用了范围比较，也还是检查了所有文件。使用简单索引对此场景更合适，因为它不采用提前的裁剪操作，而是直接和所有文件的所需字段连接。如果额外的运维成本可以接受的话，也可以采用HBase索引，其对这些表能提供更加优越的查询效率。

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

2.4、Table Types& Queries

Hudi表类型定义了如何在DFS上对数据进行索引和布局，以及如何在此类组织上实现上述操作和时间轴活动（即如何写入数据）。同样，查询类型定义了底层数据如何暴露给查询（即如何读取数据）。

Table Type Supported Query types
Copy on Write （写时复制） 快照查询+增量查询
Merge on Read （读时合并） 快照查询+增量查询+读取优化查询（近实时）

Table Types:

Copy on Write:使用列式存储来存储数据（例如：parquet）,通过在写入期间执行同步合并来简单地更新和重现文件
Merge on Read:使用列式存储（parquet）+行式文件（arvo）组合存储数据。更新记录到增量文件中，然后进行同步或异步压缩来生成新版本的列式文件。

下面总结了两种表类型之间的权衡
权衡 CopyOnWrite MergeOnRead
数据延迟 高 低
查询延迟 低 高
Update(I/O) 更新成本 高（重写整个Parquet文件） 低（追加到增量日志）
Parquet File Size 低（更新成本I/O高） 较大（低更新成本）
Write Amplification（WA写入放大） 大 低（取决于压缩策略）

Query Types:

Snapshot Queries:快照查询，在此视图上的查询将看到某个提交和压缩操作的最新快照。对于merge on read的表，它通过即时合并最新文件切片的基本文件和增量文件来展示近乎实时的数据（几分钟）。对于copy on write的表，它提供了对现有parquet表的直接替代，同时提供了upsert/delete和其他写入功能。
Incremental Queries:增量查询，该视图智能看到从某个提交/压缩写入数据集的新数据。该视图有效地提供了chang stream,来支持增量视图
Read Optimized Queries:读优化视图，在此视图上的查询将查看到给定提交或压缩操作中的最新快照。该视图将最新文件切片的列暴露个查询，并保证与非hudi列式数据集相比，具有相同列式查询功能。

下面总结了两种查询的权衡：
权衡 Snapshot Read Optimized
数据延迟 低 高
查询延迟 高（合并列式基础文件+行式增量日志文件） 低(原始列式数据)
2.5、Copy on Write Table

Copy on Write表中的文件切片仅包含基本/列文件，并且每次提交都会生成新版本的基本文件。换句话说，每次提交操作都会被压缩，以便存储列式数据，因此Write Amplification写入放大非常高（即使只有一个字节的数据被提交修改，我们也需要重写整个列数据文件），而读取数据成本则没有增加，所以这种表适合于做分析工作，读取密集型的操作。

下图说明了copy on write的表是如何工作的


随着数据被写入，对现有文件组的更新会为该文件组生成一个带有提交即时间标记的新切片，而插入分配一个新文件组并写入该文件组第一个切片。这些切片和提交即时时间在上图用同一颜色标识。针对图上右侧sql查询，首先检查时间轴上的最新提交并过滤掉之前的旧数据（根据时间查询最新数据），如上图所示粉色数据在10:10被提交，第一次查询是在10:10之前，所以出现不到粉色数据，第二次查询时间在10:10之后，可以查询到粉色数据（以被提交的数据）。

Copy on Write表从根本上改进表的管理方式

在原有文件上进行自动更新数据，而不是重新刷新整个表/分区
能够只读取修改部分的数据，而不是浪费查询无效数据
严格控制文件大小来保证查询性能（小文件会显著降低查询性能）

2.6、Merge on Read Table

Merge on Read表是copy on write的超集，它仍然支持通过仅向用户公开最新的文件切片中的基本/列来对表进行查询优化。用户每次对表文件的upsert操作都会以增量日志的形式进行存储，增量日志会对应每个文件最新的ID来帮助用户完成快照查询。因此这种表类型，能够智能平衡读取和写放大（wa）,提供近乎实时的数据。这种表最重要的是压缩器，它用来选择将对应增量日志数据压缩到表的基本文件中，来保持查询时的性能（较大的增量日志文件会影响合并时间和查询时间）

下图说明了该表的工作原理，并显示两种查询类型：快照查询和读取优化查询


如上图所示，现在每一分钟提交一次，这种操作是在别的表里（copy on write table）无法做到的
现在有一个增量日志文件，它保存对基本列文件中记录的传入更新（对表的修改），在图中，增量日志文件包含从10:05到10:10的所有数据。基本列文件仍然使用commit来进行版本控制，因此如果只看基本列文件，那么表的表的布局就像copy on write表一样。
定期压缩过程会协调增量日志文件和基本列文件进行合并，并生成新版本的基本列文件，就如图中10：05所发生的情况一样。
查询表的方式有两种，Read Optimized query和Snapshot query,取决于我们选择是要查询性能还是数据新鲜度
如上图所示，Read Optimized query查询不到10:05之后的数据（查询不到增量日志里的数据），而Snapshot query则可以查询到全量数据（基本列数据+行式的增量日志数据）。
压缩触发是解决所有难题的关键，通过实施压缩策略，会快速缩新分区数据，来保证用户使用Read Optimized query可以查询到X分钟内的数据

Merge on Read Table是直接在DFS上启用近实时（near real-time）处理，而不是将数据复制到外部专用系统中。该表还有些次要的好处，例如通过避免数据的同步合并来减少写入放大（WA）
2.7、Compaction

没有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和Sparkstreaming的writer都可以apply异步的compaction策略，按照间隔commits数或者时间来触发compaction任务，在独立的pipeline中执行。
3、基础用例
3.1、近实时读取

Hudi在各种读取数据方面也有很多好处，Hudi在DFS分布式存储系统上存储数据强制执行了最小文件大小，这样有助于解决HDFS和存储上的小文件问题，显著的提升了查询性能。并且Hudi每次提交数据的方式都是原子性的，这样也避免了查询时受到部分写入的影响。

将外部各种OLTP的数据源（比如日志数据、数据库、外部源）写入到Hudi中也是一个常见的问题，Hudi存储这些数据，原始数据层的构建也是非常关键。

对应RDBMS这种关系型数据库的数据写入，Hudi提供了Upserts操作来提供增量修改或新增，而不是采用昂贵的且低效的批量加载。使用Debezium或Kafka Connect或Sqoop等工具将数据导入到Hudi对应在DFS上的表是非常常见的一种方案。对于像NoSql这样的数据库（Cassandra / Voldemort / HBase）即使中等规模的数据量会存储十亿行左右，所以使用完全批量加载的方式根本不行，如果读取要跟上数据的高更新的变化量，则需要更有效的方法。

即使对于像Kafka这样的不可变数据源，通常也需要根据DFS上存储的内容对传入事件进行重复数据的删除。Hudi通过使用不同类型的索引来快速解决这个问题。

所有的一切都可以通过Hudi DeltaStreamer工具无缝实现，该工具Hudi官方一直在尝试增加更多数据源，该工具提供一个连续模式，可以异步地自我管理集群/压缩，而不会阻止数据的写入，显著提高数据新鲜度。

hudi0.8对接flink还存在bug。使用hudi0.9可以解决

Hudi+Debezium+flink+spark同步方案：

Apache Hudi在Linkflow构建实时数据湖的生产实践https://mp.weixin.qq.com/s?__biz=MzIyMzQ0NjA0MQ==&mid=2247485857&idx=1&sn=ef0329a4eccdff7998d5a9a45a773e77&chksm=e81f5cd7df68d5c1826ab76a51a0dbe6f04b63b6b5062247ac297c2f6b860f4cd2906578c3be&scene=178&cur_album_id=1608246604526911489#rd
3.2、数据删除

Hudi还提供了删除存储在数据中的数据的能力，更重要的是提供了处理大量写入放大(wa)的有效方法，这些通过Merge On Read 表类型基于user_id(任何辅助键)随件删除产生的结果。Hudi可以基于日志提供优雅的并发控制，保证数据的写入和读取可以持续发生，因为后台压缩作业分摊数据的重写和强制删除所需要的成本。

Hudi还解锁了数据聚类等特殊功能，允许用户优化数据布局来进行删除。具体来说，用户可以基于user_id（辅助键）对旧的事件日志数据进行聚类，这样评估数据需要删除的数据就可以快速的定位，对于分区则在时间戳上进行了聚类优化，提高查询性能。
3.3、分析和存储

数据的存储和分析一般我们分为两类数据，实时处理和离线批量处理。通常实时处理由Druid、Memsql或clickhouse提供支持，并且有kafka或pulsar提供服务，这种模型非常昂贵。

如果数据会在很晚之后才被写入到数据湖中，那么就需要进行批处理，Hudi也提供对Persto/Spark Sql等交互时Sql引擎的访问，这些引擎可以轻松横向扩展，并在几秒钟诶返回查询结果。

与实时数据集市相比，通过将数据新鲜度缩短到几分钟，Hudi可以为大量数据应用程序提供更高效的替代方案。
3.4、增量处理管道

在传统数据仓库中，整个数仓的工作流可能会发生数据延迟的问题，比如上游工作流U每小时会在Hive建立分区，每小时结束时使用该处理时间作为event time，提供1小时的数据有效新鲜度，然后下游工作流D在U完成后立即启动，并在接下来一小时内进行自己的处理，将有效延迟增加到了2小时。这样的例子忽略了数据迟到现象，即processing_time和event time并不是同一时间而是分开的，比如移动设备或者传感器间歇性的连接所造成，这种现象并不是异常而是常态,那么在这种情况下，保证正确性的唯一补救措施是每小时一遍一遍得重新处理那些迟到的数据，这样可能会严重危害到整个生态系统的效率。

那么Hudi也提供了解决方案，它提供一种以记录粒度（不是目录/分区）从上有Hudi表中消费数据（包括后期迟到数据）的方法，可以有效地更新/协调后期数据到达，上下游两张表的调度可以更频繁，例如15分钟，并可以在下游表提供30分钟的端延迟。为了实现这一目标，Hudi采纳了spark streaming、Kafka、Flink、Oracle Streaming等技术框架的概念。

具体可以查看链接：

Uber’s case for incremental processing on Hadoop – O’Reilly
————————————————
版权声明：本文为CSDN博主「Yuan_CSDF」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/Yuan_CSDF/article/details/122032205

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack