Hudi（16）：Hudi集成Flink之读取方式-Toy模板网

这篇具有很好参考价值的文章主要介绍了Hudi（16）：Hudi集成Flink之读取方式。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

0. 相关文章链接

1. 流读（Streaming Query）

2. 增量读取（Incremental Query）

3. 限流

0. 相关文章链接

Hudi文章汇总

1. 流读（Streaming Query）

当前表默认是快照读取，即读取最新的全量快照数据并一次性返回。通过参数read.streaming.enabled 参数开启流读模式，通过 read.start-commit 参数指定起始消费位置，支持指定 earliest 从最早消费。

WITH参数：

名称	Required	默认值	说明
read.streaming.enabled	false	false	设置 true 开启流读模式
read.start-commit	false	最新 commit	指定 'yyyyMMddHHmmss' 格式的起始 commit（闭区间）
read.streaming.skip_compaction	false	false	流读时是否跳过 compaction 的 commits，跳过 compaction 有两个用途： 1）避免 upsert 语义下重复消费 (compaction 的 instant 为重复数据，如果不跳过，有小概率会重复消费) 2) changelog 模式下保证语义正确性 0.11 开始，以上两个问题已经通过保留 compaction 的 instant time 修复
clean.retain_commits	false	10	cleaner 最多保留的历史 commits 数，大于此数量的历史 commits 会被清理掉，changelog 模式下，这个参数可以控制 changelog 的保留时间，例如 checkpoint 周期为 5 分钟一次，默认最少保留 50 分钟的时间。

注意：当参数 read.streaming.skip_compaction 打开并且 streaming reader 消费落后于clean.retain_commits 数时，流读可能会丢失数据。从 0.11 开始，compaction 不会再变更 record 的 instant time，因此理论上数据不会再重复消费，但是还是会重复读取并丢弃，因此额外的开销还是无法避免，对性能有要求的话还是可以开启此参数。

CREATE TABLE t5(
uuid VARCHAR(20) PRIMARY KEY NOT ENFORCED,
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
`partition` VARCHAR(20)
) WITH (
'connector' = 'hudi',
'path' = 'hdfs://hadoop1:8020/tmp/hudi_flink/t5',
'table.type' = 'MERGE_ON_READ',
'read.streaming.enabled' = 'true',
'read.streaming.check-interval' = '4' -- 默认60s
);

insert into t5 select * from sourceT;

select * from t5;

2. 增量读取（Incremental Query）

Hudi从 0.10.0 开始支持增量读取。如果有增量读取 batch 数据的需求，增量读取包含三种场景：

Stream 增量消费，通过参数 read.start-commit 指定起始消费位置；
Batch 增量消费，通过参数 read.start-commit 指定起始消费位置，通过参数 read.end-commit 指定结束消费位置，区间为闭区间，即包含起始、结束的 commit；
TimeTravel：Batch 消费某个时间点的数据：通过参数 read.end-commit 指定结束消费位置即可（由于起始位置默认从最新，所以无需重复声明）；

WITH 参数：

名称	Required	默认值	说明
read.start-commit	false	默认从最新 commit	支持 earliest 从最早消费
read.end-commit	false	默认到最新 commit