大数据Hadoop之——新一代流式数据湖平台 Apache Hudi

一、概述

Hudi（Hadoop Upserts Deletes and Incrementals），简称Hudi，是一个流式数据湖平台，支持对海量数据快速更新，内置表格式，支持事务的存储层、 一系列表服务、数据服务(开箱即用的摄取工具)以及完善的运维监控工具，它可以以极低的延迟将数据快速存储到HDFS或云存储（S3）的工具，最主要的特点支持记录级别的插入更新（Upsert）和删除，同时还支持增量查询。

• Apache Hudl本身不存储数据，仅仅管理数据，借助外部存储引擎存储数据，比如HDFS；
• 此外，Apache Hudi也不分析数据，需要使用计算分析引擎，查询和保存数据,比如Spark或Flink；
• 使用Hudi时，加载jar包，底层调用API,所以需要依据使用大数据框架版本，编译Hudi源码，获取对应依赖jar包。

GitHub地址：https://github.com/apache/hudi

官方文档：https://hudi.apache.org/cn/docs/overview

上图从下到上，由左向右看

• hudi 底层的数据可以存储到hdfs、s3、azure、alluxio 等存储。
• hudi 可以使用spark/flink 计算引擎来消费 kafka、pulsar 等消息队列的数据，而这些数据可能来源于 app 或者微服务的业务数据、日志数据，也可以是 mysql 等数据库的 binlog 日志数据。
• spark/hudi 首先将这些数据处理为 hudi 格式的 row tables （原始表），然后这张原始表可以被 Incremental ETL (增量处理)生成一张 hudi 格式的 derived tables 派生表。
• hudi 支持的查询引擎有：trino、hive、impala、spark、presto 等。
• 支持 spark、flink、map-reduce 等计算引擎继续对 hudi 的数据进行再次加工处理。

二、Hudi 架构

1. 通过DeltaStreammer、Flink、Spark等工具，将数据摄取到数据湖存储，可使用HDFS作为数据湖的数据存储；
2. 基于HDFS可以构建Hudi的数据湖；
3. Hudi提供统一的访问Spark数据源和Flink数据源；
4. 外部通过不同引擎，如：Spark、Flink、Presto、Hive、Impala、Aliyun DLA、AWS Redshit访问接口；

三、Hudi的表格式

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 的列存结构。

1）Copy on Write（写时复制）

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

• 优点：读取时，只读取对应分区的一个数据文件即可，较为高效；
• 缺点：数据写入的时候，需要复制一个先前的副本再在其基础上生成新的数据文件，这个过程比较耗时。

2）Merge On Read（读时合并）

简称MOR，新插入的数据存储在delta log 中，定期再将delta log合并进行parquet数据文件。读取数据时，会将delta log跟老的数据文件做merge，得到完整的数据返回。下图演示了MOR的两种数据读写方式。

• 优点：由于写入数据先写delta log，且delta log较小，所以写入成本较低；
• 缺点：需要定期合并整理compact，否则碎片文件较多。读取性能较差，因为需要将delta log和老数据文件合并

3）COW vs MOR

• COW表，用户在 snapshot 读取的时候会扫描所有最新的 FileSlice 下的 base file。
• MOR表，在 READ OPTIMIZED 模式下，只会读最近的经过 compaction 的 commit。

四、元数据表（Metadata Table）

• Apache Hudi元数据表可以显著提高查询的读/写性能。元数据表的主要目的是消除“列出文件”操作的要求。
• 读取和写入数据时，将执行文件列表操作以获取文件系统的当前视图。当数据集很大时，列出所有文件可能是性能瓶颈，但更重要的是，对于AWS S3等云存储系统，由于某些请求限制，大量的文件列出请求有时会导致节流。相反，元数据表将主动维护文件列表，并消除递归文件列表操作的需要。

五、索引（Indexing）

Hudi通过索引机制将给定的hoodie键（记录键+分区路径）一致地映射到文件id，从而提供高效的升级。一旦将记录的第一个版本写入文件，记录键和文件组/文件id之间的映射就不会改变。简而言之，映射的文件组包含一组记录的所有版本。

目前，Hudi支持以下索引类型：

• Bloom索引（默认）：使用由记录键构建的Bloom过滤器，也可以使用记录键范围修剪候选文件。

• 简单索引：根据从存储上的表中提取的键，对传入的更新/删除记录执行精简联接。

• HBase索引：管理外部Apache HBase表中的索引映射。

• 自定义索引：当然也可以扩展这个公共API来实现自定义索引。

六、查询类型（Query Type）

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

1）Snapshot Queries(快照查询)

• 查询查看给定提交或压缩操作时表的最新快照。在合并读取表的情况下，它通过动态合并最新文件切片的基本文件和增量文件来公开接近实时的数据（几分钟）。
• 对于随写复制表，它提供了现有拼花桌的插入式替换，同时提供了upsert/delete和其他写入端功能。

2）Incremental Queries(增量查询)

• 在给定的提交/压缩之后，查询只会看到写入表的新数据。这有效地提供了更改流以启用增量数据管道。
• 可查看自给定commit/delta commit即时操作依赖新写入的数据，有效地提供变更流来启用增量数据管道。

3）Read Optimized Queries(读优化查询)

• 查询查看给定提交/压缩操作时表的最新快照。仅显示最新文件切片中的基/列文件，并确保与非hudi列表相比具有相同的列查询性能。
• 读优化查询和快照查询相同仅访问基本文件，提供给定文件片自上次执行压缩操作以来的数据。通常查询数据的最新程度的保证取决于压缩策略。

七、计算模型

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

1）批式模型(Batch)

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

• 延迟：小时级延迟或者天级别延迟。这里的延迟不单单指的是定时任务的时间，在数据架构里，这里的延迟时间通常是定时任务间隔时间+一系列依赖任务的计算时间+数据平台最终可以展示结果的时间。数据量大、逻辑复杂的情况下，小时任务计算的数据通常真正延迟的时间是2-3小时。
• 数据完整度：数据较完整。以处理时间为例，小时级别的任务，通常计算的原始数据已经包含了小时内的所有数据，所以得到的数据相对较完整。但如果业务需求是事件时间，这里涉及到终端的一些延迟上报机制，在这里，批式计算任务就很难派上用场。
• 成本：成本很低。只有在做任务计算时，才会占用资源，如果不做任务计算，可以将这部分批式计算资源出让给在线业务使用。从另一个角度来说成本是挺高的，如原始数据做了一些增删改查，数据晚到的情况，那么批式任务是要全量重新计算。

2）流式模型(Stream)

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

• 延迟：很短，甚至是实时。
• 数据完整度：较差。因为流式引擎不会等到所有数据到齐之后再开始计算，所以有一个watermark的概念，当数据的时间小于watermark时，就会被丢弃，这样是无法对数据完整度有一个绝对的保障。在互联网场景中，流式模型主要用于活动时的数据大盘展示，对数据的完整度要求并不算很高。在大部分场景中，用户需要开发两个程序，一是流式数据生产流式结果，而是批式计算人物，用于次日修复实时结果。
• 成本：很高。因为流式任务时常驻的，并且对于多流join的场景，通常要借助内存或者数据库来做state的存储，不管是序列化开销，还是和外部组件交互产生的额外IO，在大数据量下都是不容忽视的。

3）增量模型(Incremental)

针对批式和流式的优缺点，Uber提出了增量模型(Incremental Mode)，相对批式来讲，更加实时；相对流式而言，更加经济。 增量模型，简单来讲，就是一mini batch的形式来跑准实时任务。hudi在增量模型中支持了两个最重要的特性：

• Upsert：这个主要是解决批式模型中，数据不能插入、更新的问题，有了这个特性，可以往Hive中写入增量数据，而不是每次进行完全的覆盖。（hudi自身维护了key-file的映射，所以当upsert时很容易找到key对应的文件）
• Incremental Query：增量查询，减少计算的原始数据量。以uber中司机和乘客的数据流join为例，每次抓取两条数据流中的增量数据进行批式的join即可，相比流式数据而言，成本要降低几个数量级。

八、数据仓库 VS 数据湖

1）数据类型

• 结构化数据——来自关系型数据库中的行和列。
• 半结构化数据——如CSV、日志、XML、JSON等。
• 非结构化数据——如email、文档、PDF等。
• 二进制数据——如图像、音频、视频等。

2）数据仓库与数据湖的区别

• 数据仓库可以理解为是一个优化的数据库，用户分析来自事物系统和业务线应用程序的关系型数据（结构化数据和半结构化数据）。
• 数据湖可以理解存储来自业务应用程序的关系型数据（结构化数据），以及来自移动应用程序、IOT设备和社交媒体的非关系型数据（非结构化数据）等所有类型数据。

3）湖仓一体化

• Data Lakehouse （湖仓一体）是新出现的一种数据架构,它同时吸收了数据仓库和数据湖的优势，数据分析师和数据科学家可以在同一个数据存储中对数据进行操作，同时它也能为公司进行数据治理带来更多的便利性。
• LakeHouse使用新的系统设计：直接在用于数据湖的低成本存储上实现与数据仓库中类似的数据结构和数据管理功能。

九、源码编译

wget https://dlcdn.apache.org/hudi/0.12.0/hudi-0.12.0.src.tgz
tar -xf hudi-0.12.0.src.tgz ; cd cd hudi-0.12.0
# mvn clean package -DskipTests
mvn clean package -DskipTests -DskipITs -Dspark3.2 -Dscala-2.12
1
2
3
4

编译好的Hudi 包下载地址：

链接：https://pan.baidu.com/s/15qKm1kW1RRtbyFT53RoeGA?pwd=ihhb
提取码：ihhb

新一代流式数据湖平台 Apache Hudi介绍与源码编译就先到这里了，有疑问的小伙伴欢迎给我留言，后续会持续更新【云原生+大数据】相关的文章，请小伙伴耐心等待~