Hudi集成Spark之spark-sql方式
啟動spark-sql
# 啟動spark-sql之前需要先啟動Hive的Metastore
nohup hive --service metastore &
#針對Spark 3.2
spark-sql \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
# 如果沒有配置hive環(huán)境變量，手動拷貝hive-site.xml到spark的conf下
創(chuàng)建表
建表參數(shù)：
| 參數(shù)名 | 默認值 | 說明 |
|---|---|---|
| primaryKey | uuid | 表的主鍵名，多個字段用逗號分隔。同 hoodie.datasource.write.recordkey.field |
| preCombineField | 表的預(yù)合并字段。同 hoodie.datasource.write.precombine.field | |
| type | cow | 創(chuàng)建的表類型： type = ‘cow’ type = 'mor’同 hoodie.datasource.write.table.type |
（1）創(chuàng)建非分區(qū)表
創(chuàng)建一個 cow 表，默認 primaryKey ‘uuid’，不提供 preCombineField
create database spark_hudi;
use spark_hudi;
create table hudi_cow_nonpcf_tbl (
uuid int,
name string,
price double
) using hudi;
（2）創(chuàng)建一個 mor 非分區(qū)表
create table hudi_mor_tbl (
id int,
name string,
price double,
ts bigint
) using hudi
tblproperties (
type = 'mor',
primaryKey = 'id',
preCombineField = 'ts'
);
（3）創(chuàng)建分區(qū)表
創(chuàng)建一個 cow 分區(qū)外部表，指定 primaryKey 和 preCombineField。此刻數(shù)據(jù)在hdfs上
create table hudi_cow_pt_tbl (
id bigint,
name string,
ts bigint,
dt string,
hh string
) using hudi
tblproperties (
type = 'cow',
primaryKey = 'id',
preCombineField = 'ts'
)
partitioned by (dt, hh)
location '/opt/hudi/hudi_cow_pt_tbl';
（4）在已有的 hudi 表上創(chuàng)建新表，不需要指定模式和非分區(qū)列（如果存在）之外的任何屬性，Hudi 可以自動識別模式和配置。
-
非分區(qū)表
create table hudi_existing_tbl0 using hudi location 'file:///opt/datas/hudi/dataframe_hudi_nonpt_table'; -
分區(qū)表
create table hudi_existing_tbl1 using hudi partitioned by (dt, hh) location 'file:///opt/datas/dataframe_hudi_pt_table';
（5）通過 CTAS (Create Table As Select)建表為了提高向 hudi 表加載數(shù)據(jù)的性能，CTAS 使用批量插入作為寫操作。
-
通過 CTAS 創(chuàng)建 cow 非分區(qū)表，不指定 preCombineField
create table hudi_ctas_cow_nonpcf_tbl using hudi tblproperties (primaryKey = 'id') as select 1 as id, 'a1' as name, 10 as price; -
通過 CTAS 創(chuàng)建 cow 分區(qū)表，指定 preCombineField
create table hudi_ctas_cow_pt_tbl using hudi tblproperties (type = 'cow', primaryKey = 'id', preCombineField = 'ts') partitioned by (dt) as select 1 as id, 'a1' as name, 10 as price, 1000 as ts, '2021-12-01' as dt; -
通過 CTAS 從其他表加載數(shù)據(jù)
# 創(chuàng)建內(nèi)部表 create table parquet_mngd using parquet location 'file:///opt/datas/parquet_dataset/*.parquet'; # 通過 CTAS 加載數(shù)據(jù) create table hudi_ctas_cow_pt_tbl2 using hudi location 'file://opt/datas/hudi/hudi_tbl/' options ( type = 'cow', primaryKey = 'id', preCombineField = 'ts' ) partitioned by (datestr) as select * from parquet_mngd;
插入數(shù)據(jù)
默認情況下，如果提供了 preCombineKey，則 insert into 的寫操作類型為 upsert，否則使用 insert
（1）向非分區(qū)表插入數(shù)據(jù)
insert into hudi_cow_nonpcf_tbl select 1, 'a1', 20;
insert into hudi_mor_tbl select 1, 'a1', 20, 1000;
（2）向分區(qū)表動態(tài)分區(qū)插入數(shù)據(jù)
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;
（3）向分區(qū)表靜態(tài)分區(qū)插入數(shù)據(jù)
insert into hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='11') select 2, 'a2', 1000;
（4）使用 bulk_insert 插入數(shù)據(jù)
hudi 支持使用 bulk_insert 作為寫操作的類型，只需要設(shè)置兩個配置：hoodie.sql.bulk.insert.enable 和 hoodie.sql.insert.mode。
-- 向指定 preCombineKey 的表插入數(shù)據(jù)，則寫操作為 upsert
insert into hudi_mor_tbl select 1, 'a1_1', 20, 1001;
select id, name, price, ts from hudi_mor_tbl;
1 a1_1 20.0 1001
-- 向指定 preCombineKey 的表插入數(shù)據(jù)，指定寫操作為 bulk_insert
set hoodie.sql.bulk.insert.enable=true;
set hoodie.sql.insert.mode=non-strict;
insert into hudi_mor_tbl select 1, 'a1_2', 20, 1002;
select id, name, price, ts from hudi_mor_tbl;
1 a1_1 20.0 1001
1 a1_2 20.0 1002
查詢數(shù)據(jù)
（1）查詢
select fare, begin_lon, begin_lat, ts from hudi_trips_snapshot where fare > 20.0;
（2）時間旅行查詢
Hudi 從 0.9.0 開始就支持時間旅行查詢。Spark SQL 方式要求 Spark 版本 3.2 及以上。
-- 關(guān)閉前面開啟的 bulk_insert
set hoodie.sql.bulk.insert.enable=false;
-- 數(shù)據(jù)寫入到hdfs上
create table hudi_cow_pt_tbl1 (
id bigint,
name string,
ts bigint,
dt string,
hh string
) using hudi
tblproperties (
type = 'cow',
primaryKey = 'id',
preCombineField = 'ts'
)
partitioned by (dt, hh)
location '/opt/datas/hudi/hudi_cow_pt_tbl1';
-- 插入一條 id 為 1 的數(shù)據(jù)
insert into hudi_cow_pt_tbl1 select 1, 'a0', 1000, '2021-12-09', '10';
select * from hudi_cow_pt_tbl1;
-- 修改 id 為 1 的數(shù)據(jù)
insert into hudi_cow_pt_tbl1 select 1, 'a1', 1001, '2021-12-09', '10';
select * from hudi_cow_pt_tbl1;
-- 基于第一次提交時間進行時間旅行
select * from hudi_cow_pt_tbl1 timestamp as of '20220307091628793' where id = 1;
-- 其他時間格式的時間旅行寫法
select * from hudi_cow_pt_tbl1 timestamp as of '2022-03-07 09:16:28.100' where id = 1;
select * from hudi_cow_pt_tbl1 timestamp as of '2022-03-08' where id = 1;
更新數(shù)據(jù)
（1）update
更新操作需要指定 preCombineField。
-
語法
UPDATE tableIdentifier SET column = EXPRESSION(,column = EXPRESSION) [ WHERE boolExpression] -
執(zhí)行更新
update hudi_mor_tbl set price = price * 2, ts = 1111 where id = 1; update hudi_cow_pt_tbl1 set name = 'a1_1', ts = 1001 where id = 1; -- update using non-PK field update hudi_cow_pt_tbl1 set ts = 1111 where name = 'a1_1';
（2）MergeInto
-
語法
MERGE INTO tableIdentifier AS target_alias USING (sub_query | tableIdentifier) AS source_alias ON <merge_condition> [ WHEN MATCHED [ AND <condition> ] THEN <matched_action> ] [ WHEN MATCHED [ AND <condition> ] THEN <matched_action> ] [ WHEN NOT MATCHED [ AND <condition> ] THEN <not_matched_action> ] <merge_condition> =A equal bool condition <matched_action> = DELETE | UPDATE SET * | UPDATE SET column1 = expression1 [, column2 = expression2 ...] <not_matched_action> = INSERT * | INSERT (column1 [, column2 ...]) VALUES (value1 [, value2 ...])可以看作是一個join操作。
-
執(zhí)行案例
執(zhí)行前開啟hive的hiveservice2
[root@hadoop102 bin]# ./hiveserver2 start-- 1、準備 source 表：非分區(qū)的 hudi 表，插入數(shù)據(jù) create table merge_source (id int, name string, price double, ts bigint) using hudi tblproperties (primaryKey = 'id', preCombineField = 'ts'); insert into merge_source values (1, "old_a1", 22.22, 2900), (2, "new_a2", 33.33, 2000), (3, "new_a3", 44.44, 2000); merge into hudi_mor_tbl as target using merge_source as source on target.id = source.id when matched then update set * when not matched then insert *; -- 2、準備 source 表：分區(qū)的 parquet 表，插入數(shù)據(jù) create table merge_source2 (id int, name string, flag string, dt string, hh string) using parquet; insert into merge_source2 values (1, "new_a1", 'update', '2021-12- 09', '10'), (2, "new_a2", 'delete', '2021-12-09', '11'), (3, "new_a3", 'insert', '2021-12-09', '12'); merge into hudi_cow_pt_tbl1 as target using ( select id, name, '2000' as ts, flag, dt, hh from merge_source2 ) source on target.id = source.id when matched and flag != 'delete' then update set id = source.id, name = source.name, ts = source.ts, dt = source.dt, hh = source.hh when matched and flag = 'delete' then delete when not matched then insert (id, name, ts, dt, hh) values(source.id, source.name, source.ts, source.dt, source.hh);
mergeInto會發(fā)生的報錯：
Could not sync using the meta sync class org.apache.hudi.hive.HiveSyncTool
java.sql.SQLException: Could not open client transport with JDBC Uri: jdbc:hive2://localhost:10000: Failed to open new session: java.lang.RuntimeException: org.apache.hadoop.security.AccessControlException: Permission denied: user=hive, access=EXECUTE, inode="/tmp":root:supergroup:drwxrwx---
解決方案：https://blog.csdn.net/weixin_45417821/article/details/128651942
刪除數(shù)據(jù)
語法：
DELETE FROM tableIdentifier [ WHERE BOOL_EXPRESSION]
案例：
delete from hudi_cow_nonpcf_tbl where uuid = 1;
delete from hudi_mor_tbl where id % 2 = 0;
-- 使用非主鍵字段刪除
delete from hudi_cow_pt_tbl1 where name = 'a1_1';
覆蓋數(shù)據(jù)
- 使用 INSERT_OVERWRITE 類型的寫操作覆蓋分區(qū)表
- 使用 INSERT_OVERWRITE_TABLE 類型的寫操作插入覆蓋非分區(qū)表或分區(qū)表（動態(tài)分區(qū)）
（1）insert overwrite 非分區(qū)表
insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
insert overwrite hudi_cow_nonpcf_tbl select 99, 'a99', 20.0;
（2）通過動態(tài)分區(qū) insert overwrite table 到分區(qū)表
insert overwrite table hudi_cow_pt_tbl1 select 10, 'a10', 1100, '2021-12-09', '11';
（3）通過靜態(tài)分區(qū) insert overwrite 分區(qū)表
insert overwrite hudi_cow_pt_tbl1 partition(dt = '2021-12-09', hh='12') select 13, 'a13', 1100;
修改表結(jié)構(gòu)（Alter Table）
語法：
-- Alter table name
ALTER TABLE oldTableName RENAME TO newTableName
-- Alter table add columns
ALTER TABLE tableIdentifier ADD COLUMNS(colAndType (,colAndType)*)
-- Alter table column type
ALTER TABLE tableIdentifier CHANGE COLUMN colName colName colType
-- Alter table properties
ALTER TABLE tableIdentifier SET TBLPROPERTIES (key = 'value')
案例：
--rename to:
ALTER TABLE hudi_cow_nonpcf_tbl RENAME TO hudi_cow_nonpcf_tbl2;
--add column:
ALTER TABLE hudi_cow_nonpcf_tbl2 add columns(remark string);
--change column:
ALTER TABLE hudi_cow_nonpcf_tbl2 change column uuid uuid int;
--set properties;
alter table hudi_cow_nonpcf_tbl2 set tblproperties (hoodie.keep.max.commits = '10');
修改分區(qū)
語法：
-- Drop Partition
ALTER TABLE tableIdentifier DROP PARTITION ( partition_col_name = partition_col_val [ , ... ] )
-- Show Partitions
SHOW PARTITIONS tableIdentifier
案例：
--show partition:
show partitions hudi_cow_pt_tbl1;
--drop partition：
alter table hudi_cow_pt_tbl1 drop partition (dt='2021-12-09', hh='10');
注意：show partition 結(jié)果是基于文件系統(tǒng)表路徑的。刪除整個分區(qū)數(shù)據(jù)或直接刪除某個分區(qū)目錄并不精確。
存儲過程（Procedures）
語法：
--Call procedure by positional arguments
CALL system.procedure_name(arg_1, arg_2, ... arg_n)
--Call procedure by named arguments
CALL system.procedure_name(arg_name_2 => arg_2, arg_name_1 =>
arg_1, ... arg_name_n => arg_n)
案例：文章來源：http://www.zghlxwxcb.cn/news/detail-447928.html
可用的存儲過程：https://hudi.apache.org/docs/procedures/文章來源地址http://www.zghlxwxcb.cn/news/detail-447928.html
--show commit's info
call show_commits(table => 'hudi_cow_pt_tbl1', limit => 10);
到了這里，關(guān)于Hudi-集成Spark之spark-sql方式的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
