Hudi 是一个通用的大数据存储系统，主要特性：
·摄取和查询引擎之间的快照隔离，包括 Hive、Presto 和 Spark。
·支持回滚和存储点，可以恢复数据集。
·自动管理文件大小和布局，以优化查询性能准实时摄取，为查询提供最新数据。
·实时数据和列数据的异步压缩。
一、时间轴
Hudi 维护一条包含在不同的即时时间所有对数据集操作的时间轴，从而提供了从不同时间点出发得到不同的视图下的数据集。
Hudi 即时包含以下组件：
·操作类型：对数据集执行的操作类型。
·即时时间：即时时间通常是一个时间戳（例如20190117010349），该时间戳按操作开始时间的顺序单调增加。
·状态：即时的状态。
二、文件组织
Hudi 将 DFS 上的数据集组织到基本路径下的目录结构中。数据集分为多个分区，这些分区是包含该分区的数据文件的文件夹，这与 Hive 表非常相似。
每个分区被相对于基本路径的特定分区路径区分开来。在每个分区内，文件被组织为文件组，由文件id唯一标识。每个文件组包含多个文件切片，其中每个切片包含在某个提交/压缩即时时间生成的基本列文件*.parquet以及一组日志文件*.log*，该文件包含自生成基这篇件以来对基这篇件的插入/更新。
Hudi 采用 MVCC 设计，其中压缩操作将日志和基这篇件合并以产生新的文件片，而清理操作则将未使用的/较旧的文件片删除以回收 DFS 上的空间。Hudi 通过索引机制将给定的 hoodie 键（记录键+分区路径）映射到文件组，从而提供了高效的 Upsert。
一旦将记录的第一个版本写入文件，记录键和文件组/文件id之间的映射就永远不会改变。简而言之，映射的文件组包含一组记录的所有版本。
三、存储类型
Hudi 支持以下存储类型：
·写时复制：仅使用列文件格式（例如 parquet）存储数据。通过在写入过程中执行同步合并以更新版本并重写文件。
·读时合并：更新记录到增量文件中，然后进行同步或异步压缩以生成列文件的新版本。
下表总结了这两种存储类型之间的权衡：
| 权衡 | 写时复制 | 读时合并 |
| 数据延迟 | 更高 | 更低 |
| 查询延迟 | 更低 | 更高 |
| 更新代价（I/O） | 更高（重写整个 parquet 文件） | 更低（追加到增量日志） |
| 写放大 | 更高 | 更低（取决于压缩策略） |
四、Hudi 对 EMR 底层存储支持
·HDFS
·COS
五、安装 Hudi
进入 EMR 购买页，选择产品版本为 EMR-V3.6.0，选择可选组件为 Hudi 0.13.0。各个产品版本对应的 Hudi 组件版本可以查看组件版本概览页面。
注意：Hudi 组件依赖 Hive 和 Spark 组件， 如果选择安装 Hudi 组件，需同时选择安装 Hive 和 Spark 组件。
六、使用示例
本示例将演示使用 SparkSQL 建表并写入数据，然后使用 Hive 和 Trino 进行查询分析。
七、使用 SparkSQL 读写 Hudi 表
·登录 master 节点，切换为 hadoop 用户。
·Hudi 支持使用 SparkSQL 的 HoodieSparkSessionExtension 扩展进行数据读写：
spark-sql –master yarn \\–num-executors 2 \\–executor-memory 1g \\–executor-cores 2 \\–jars /usr/local/service/hudi/hudi-bundle/hudi-spark3.3-bundle_2.12-0.13.0.jar \\–conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \\–conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \\–conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'
注意：说明：
注意：其中 –master 表示你的 master URL，–num-executors 表示 executor 数量，–executor-memory 表示 executor 的储存容量，以上参数也可以根据你的实际情况作出修改。 –jars 使用的依赖包版本在不同 EMR 版本中可能存在差异，请在 /usr/local/service/hudi/hudi-bundle 目录下查看并使用正确的依赖包。
·建表
— 创建COW非分区表spark-sql> create table hudi_cow_nonpcf_tbl ( uuid int, name string, price double) using huditblproperties ( primaryKey = 'uuid');– 创建COW分区表spark-sql> create table hudi_cow_pt_tbl ( id bigint, name string, ts bigint, dt string, hh string) using huditblproperties ( type = 'cow', primaryKey = 'id', preCombineField = 'ts' )partitioned by (dt, hh);– 创建MOR分区表spark-sql> create table hudi_mor_tbl ( id int, name string, price double, ts bigint, dt string) using huditblproperties ( type = 'mor', primaryKey = 'id', preCombineField = 'ts')partitioned by (dt);
·写入数据
— insert into non-partitioned tablespark-sql> insert into hudi_cow_nonpcf_tbl select 1, 'a1', 20;– insert dynamic partitionspark-sql> insert into hudi_cow_pt_tbl partition (dt, hh) select 1 as id, 'a1' as name, 1000 as ts, '2021-12-09' as dt, '10' as hh;– insert static partitionspark-sql> insert into hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='11') select 2, 'a2', 1000;spark-sql> insert into hudi_mor_tbl partition(dt = '2021-12-09') select 1, 'a1', 20, 1000;
·查询数据
spark-sql> select * from hudi_cow_nonpcf_tbl;20230808165859974 20230808165859974_0_1 uuid:1 f564e6c7-631c-4f32-a01d-e042f37ad6e6-0_0-21-19_20230808165859974.parquet 1 a1 20.0Time taken: 6.255 seconds, Fetched 1 row(s)spark-sql> select count(*) from hudi_mor_tbl;1Time taken: 0.955 seconds, Fetched 1 row(s)spark-sql> select name, count(*) from hudi_cow_pt_tbl group by name;a2 1a1 1Time taken: 2.049 seconds, Fetched 2 row(s)
八、使用 Hive 查询 Hudi 表
以下演示在 Hive CLI 中如何查询上述创建的 Hudi 表：
注意：说明：
注意：不同 EMR 版本中加载的 Hudi 依赖包版本可能存在不同，请在/usr/local/service/hudi/hudi-bundle目录下选择正确版本依赖包进行加载。
hive> add jar /usr/local/service/hudi/hudi-bundle/hudi-hadoop-mr-bundle-0.13.0.jar;Added [/usr/local/service/hudi/hudi-bundle/hudi-hadoop-mr-bundle-0.13.0.jar] to class pathhive> show tables;OKhudi_cow_nonpcf_tblhudi_cow_pt_tblhudi_mor_tblhudi_mor_tbl_rohudi_mor_tbl_rtTime taken: 0.023 seconds, Fetched: 5 row(s)hive> set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;hive> select * from hudi_mor_tbl_ro;OK20230808174602565 20230808174602565_0_1 id:1 dt=2021-12-09 af40667d-1dca-4163-89ca-2c48250985b2-0_0-34-1617_20230808174602565.parquet 1 a1 20.0 1000 2021-12-09Time taken: 0.159 seconds, Fetched: 1 row(s)hive> set hive.vectorized.execution.enabled=false;hive> select name, count(*) from hudi_mor_tbl_rt group by name;a1 1Time taken: 17.618 seconds, Fetched: 1 row(s)
九、使用 Trino 查询 Hudi 表
以下演示如何在 Trino 中查询上述创建的 Hudi 表。
·EMR 3.6.0之前版本：
·Trino 查询 Hudi 使用 Hive Connector，无需额外配置。
·将 /usr/local/service/hudi/hudi-bundle/hudi-hadoop-mr-bundle-$hudi_version.jar 拷贝到集群所有节点的/usr/local/service/trino/plugin/hive/路径下，并重启 Trino 服务。
·EMR-3.6.0版本及以后版本：
·需要在 Trino 组件配置管理中添加 hudi connector 配置文件。参考配置更新说明进行新增配置文件操作。其中 ${hivemetastore_uri} 需要改为集群的 Hive Metastore 服务地址，具体可以查看 Hive 组件配置文件 hive-site.xml 中的 hive.metastore.uris 配置项。
·配置完成后重启Trino服务。
connector.name=hudihive.metastore.uri=${hivemetatore_uri}
完成以上步骤后，可以在 Trino 中查询 Hudi 表：
trino:default> show tables; Table——————— hudi_cow_nonpcf_tbl hudi_cow_pt_tbl hudi_mor_tbl hudi_mor_tbl_ro hudi_mor_tbl_rt(5 rows)Query 20230808_113217_00005_r3tqk, FINISHED, 3 nodesSplits: 12 total, 12 done (100.00%)0.34 [11 rows, 359B] [32 rows/s, 1.03KB/s]trino:default> select uuid, name, price from hudi_cow_nonpcf_tbl; uuid | name | price——+——+——- 1 | a1 | 20.0(1 row)Query 20230808_114808_00001_72vwk, FINISHED, 1 nodeSplits: 5 total, 5 done (100.00%)3.48 [1 rows, 440KB] [0 rows/s, 126KB/s]trino:default> select id, name, price from hudi_mor_tbl_ro; id | name | price—-+——+——- 1 | a1 | 20.0(1 row)Query 20230808_115532_00006_72vwk, FINISHED, 1 nodeSplits: 5 total, 5 done (100.00%)1.75 [1 rows, 440KB] [0 rows/s, 251KB/s]trino:default> select name, count(*) from hudi_mor_tbl_rt group by name; name | _col1——+——- a1 | 1(1 row)Query 20230808_115728_00009_72vwk, FINISHED, 2 nodesSplits: 21 total, 21 done (100.00%)0.63 [1 rows, 440KB] [1 rows/s, 698KB/s]