?官網(wǎng)文檔
地址：https://hudi.apache.org/cn/docs/querying_data#incremental-query
參數(shù)
- read.start-commit 增量查詢開始時間 對于流讀，如果不指定該值，默認取最新的instantTime，也就是流讀默認從最新的instantTime開始讀(包含最新的)。對于批讀，如果不指定該參數(shù)，只指定read.end-commit，則實現(xiàn)時間旅行的功能，可查詢歷史記錄
- read.end-commit 增量查詢結(jié)束時間 不指定該參數(shù)則默認讀取到最新的記錄，該參數(shù)一般只適用于批讀，因為流讀一般的需求是查詢所有的增量數(shù)據(jù)
- read.streaming.enabled 是否流讀 默認false
- read.streaming.check-interval 流讀的檢查時間間隔，單位秒（s）,默認值60，也就是一分鐘查詢范圍 [BEGIN_INSTANTTIME,END_INSTANTTIME]，既包含開始時間又包含結(jié)束時間，對于默認值可參考上面的參數(shù)說明
版本
建表造數(shù)：
我這里建表造數(shù)使用Hudi Spark SQL 0.9.0,目的是為了模擬項目上用Java Client和Spark SQL創(chuàng)建的Hudi表，以驗證Hudi Flink SQL增量查詢時是否兼容舊版本的Hudi表（大家沒有這種需求的，可以使用任何方式正常造數(shù)）
查詢
- Hudi 0.13.0-SNAPSHOT
- Flink 1.14.3 （增量查詢）
- Spark 3.1.2 （主要是為了使用Call Procedures命令查看commit信息）
建表造數(shù)
-- Spark SQL Hudi 0.9.0
create table hudi.test_flink_incremental (
id int,
name string,
price double,
ts long,
dt string
) using hudi
partitioned by (dt)
options (
primaryKey = 'id',
preCombineField = 'ts',
type = 'cow'
);
insert into hudi.test_flink_incremental values (1,'a1', 10, 1000, '2022-11-25');
insert into hudi.test_flink_incremental values (2,'a2', 20, 2000, '2022-11-25');
update hudi.test_flink_incremental set name='hudi2_update' where id = 2;
insert into hudi.test_flink_incremental values (3,'a3', 30, 3000, '2022-11-26');
insert into hudi.test_flink_incremental values (4,'a4', 40, 4000, '2022-12-26');
用show_commits看一下有哪些commits（這里查詢用的是Hudi的master，因為show_commits是在0.11.0版本開始支持的，也可以通過使用hadoop命令查看.hoodie文件夾下的.commit文件）
call show_commits(table => 'hudi.test_flink_incremental');
20221205152736
20221205152723
20221205152712
20221205152702
20221205152650
Flink SQL創(chuàng)建Hudi內(nèi)存表
CREATE TABLE test_flink_incremental (
id int PRIMARY KEY NOT ENFORCED,
name VARCHAR(10),
price double,
ts bigint,
dt VARCHAR(10)
)
PARTITIONED BY (dt)
WITH (
'connector' = 'hudi',
'path' = 'hdfs://cluster1/warehouse/tablespace/managed/hive/hudi.db/test_flink_incremental'
);
建表時不指定增量查詢相關(guān)的參數(shù)，我們在查詢時動態(tài)指定，這樣比較靈活。動態(tài)指定參數(shù)方法，在查詢語句后面加上如下形式的語句
/*+
options(
'read.start-commit' = '20221205152723',
'read.end-commit'='20221205152736'
)
*/
批讀
Flink SQL讀Hudi有兩種模式：批讀和流讀。默認批讀，先看一下批讀的增量查詢
驗證是否包含起始時間和默認結(jié)束時間
select * from test_flink_incremental
/*+
options(
'read.start-commit' = '20221205152723' --起始時間對應(yīng)id=3的記錄
)
*/
結(jié)果包含起始時間，不指定結(jié)束時間默認讀到最新的數(shù)據(jù)
id name price ts dt
4 a4 40.0 4000 dt=2022-12-26
3 a3 30.0 3000 dt=2022-11-26
驗證是否包含結(jié)束時間
select * from test_flink_incremental
/*+
options(
'read.start-commit' = '20221205152712', --起始時間對應(yīng)id=2的記錄
'read.end-commit'='20221205152723' --結(jié)束時間對應(yīng)id=3的記錄
)
*/
結(jié)果包含結(jié)束時間
id name price ts dt
3 a3 30.0 3000 dt=2022-11-26
2 hudi2_update 20.0 2000 dt=2022-11-25
驗證默認開始時間
這種情況是指定結(jié)束時間，但不指定開始時間，如果都不指定，則讀表所有的最新版本的記錄。
select * from test_flink_incremental
/*+
options(
'read.end-commit'='20221205152712' --結(jié)束時間對應(yīng)id=2的更新記錄
)
*/
結(jié)果：只查詢end-commit對應(yīng)的記錄
id name price ts dt
2 hudi2_update 20.0 2000 dt=2022-11-25
時間旅行（查詢歷史記錄）
驗證是否可以查詢歷史記錄，我們更新id為2的name，更新前name為a2，更新后為hudi2_update,我們驗證一下，是否可以通過FlinkSQL查詢Hudi歷史記錄,預(yù)期結(jié)果查出id=2，name=a2
select * from test_flink_incremental
/*+
options(
'read.end-commit'='20221205152702' --結(jié)束時間對應(yīng)id=2的歷史記錄
)
*/
結(jié)果：可以正確查詢歷史記錄
id name price ts dt
2 a2 20.0 2000 dt=2022-11-25
流讀
開啟流讀的參數(shù)：
read.streaming.enabled = true
流讀不需要設(shè)置結(jié)束時間，因為一般的需求是讀所有的增量數(shù)據(jù)，我們只需要驗證開始時間就好了
驗證默認開始時間
select * from test_flink_incremental
/*+
options(
'read.streaming.enabled'='true',
'read.streaming.check-interval' = '4'
)
*/
結(jié)果：從最新的instantTime開始增量讀取，也就是默認的read.start-commit為最新的instantTime
id name price ts dt
4 a4 40.0 4000 dt=2022-12-26
驗證指定開始時間
select * from test_flink_incremental
/*+
options(
'read.streaming.enabled'='true',
'read.streaming.check-interval' = '4',
'read.start-commit' = '20221205152712'
)
*/
結(jié)果：
id name price ts dt
2 hudi2_update 20.0 2000 dt=2022-11-25
3 a3 30.0 3000 dt=2022-11-26
4 a4 40.0 4000 dt=2022-11-26
如果想第一次查詢?nèi)康臍v史數(shù)據(jù)，可以將start-commit設(shè)置的早一點，比如設(shè)置到去年：'read.start-commit' = '20211205152712'
select * from test_flink_incremental
/*+
options(
'read.streaming.enabled'='true',
'read.streaming.check-interval' = '4',
'read.start-commit' = '20211205152712'
)
*/
id name price ts dt
1 a1 10.0 1000 dt=2022-11-25
2 hudi2_update 20.0 2000 dt=2022-11-25
3 a3 30.0 3000 dt=2022-11-26
4 a4 40.0 4000 dt=2022-11-26
驗證流讀的連續(xù)性
驗證新的增量數(shù)據(jù)進來，是否可以持續(xù)消費Hudi增量數(shù)據(jù)，驗證數(shù)據(jù)的準(zhǔn)確一致性，為了方便驗證，我可以使用Flink SQL增量流讀Hudi表然后Sink到MySQL表中，最后通過讀取MySQL表中的數(shù)據(jù)驗證數(shù)據(jù)的準(zhǔn)確性
Flink SQL讀寫MySQL需要配置jar包，將flink-connector-jdbc_2.12-1.14.3.jar?放到lib?下即可,下載地址：https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc_2.12/1.14.3/flink-connector-jdbc_2.12-1.14.3.jar
先在MySQL中創(chuàng)建一張Sink表
-- MySQL
CREATE TABLE `test_sink` (
`id` int(11),
`name` text DEFAULT NULL,
`price` int(11),
`ts` int(11),
`dt` text DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
Flink中創(chuàng)建對應(yīng)的sink表
create table test_sink (
id int,
name string,
price double,
ts bigint,
dt string
) with (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://192.468.44.128:3306/hudi?useSSL=false&useUnicode=true&characterEncoding=UTF-8&characterSetResults=UTF-8',
'username' = 'root',
'password' = 'root-123',
'table-name' = 'test_sink',
'sink.buffer-flush.max-rows' = '1'
);
然后流式增量讀取Hudi表Sink Mysql
insert into test_sink
select * from test_flink_incremental
/*+
options(
'read.streaming.enabled'='true',
'read.streaming.check-interval' = '4',
'read.start-commit' = '20221205152712'
)
*/
這樣會起一個長任務(wù)，一直處于running狀態(tài)，我們可以在yarn-session界面上驗證這一點
然后先在MySQL中驗證一下歷史數(shù)據(jù)的準(zhǔn)確性
再利用Spark SQL往source表插入兩條數(shù)據(jù)
-- Spark SQL
insert into hudi.test_flink_incremental values (5,'a5', 50, 5000, '2022-12-07');
insert into hudi.test_flink_incremental values (6,'a6', 60, 6000, '2022-12-07');
我們增量讀取的間隔設(shè)置的4s，成功插入數(shù)據(jù)等待4s后，再在MySQL表中驗證一下數(shù)據(jù)
發(fā)現(xiàn)新增的數(shù)據(jù)已經(jīng)成功Sink到MySQL中了，并且數(shù)據(jù)沒有重復(fù)
最后驗證一下更新的增量數(shù)據(jù)，Spark SQL更新Hudi source表
-- Spark SQL
update hudi.test_flink_incremental set name='hudi5_update' where id = 5;
繼續(xù)驗證結(jié)果
結(jié)果是更新的增量數(shù)據(jù)也會insert到MySQL中的sink表，但是不會更新原來的數(shù)據(jù)
那如果想實現(xiàn)更新的效果呢？我們需要在MySQL和Flink的sink表中加上主鍵字段，兩者缺一不可，如下：
-- MySQL
CREATE TABLE `test_sink` (
`id` int(11),
`name` text DEFAULT NULL,
`price` int(11),
`ts` int(11),
`dt` text DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- Flink SQL
create table test_sink (
id int PRIMARY KEY NOT ENFORCED,
name string,
price double,
ts bigint,
dt string
) with (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://192.468.44.128:3306/hudi?useSSL=false&useUnicode=true&characterEncoding=UTF-8&characterSetResults=UTF-8',
'username' = 'root',
'password' = 'root-123',
'table-name' = 'test_sink',
'sink.buffer-flush.max-rows' = '1'
);
將剛才起的長任務(wù)關(guān)掉，重新執(zhí)行剛才的insert語句，先跑一下歷史數(shù)據(jù)，最后再驗證一下增量效果
-- Spark SQL
update hudi.test_flink_incremental set name='hudi6_update' where id = 6;
insert into hudi.test_flink_incremental values (7,'a7', 70, 7000, '2022-12-07');
可以看到，達到了預(yù)期效果，對于id=6的執(zhí)行更新操作，對于id=7的執(zhí)行插入操作。
