目錄

0. 相關(guān)文章鏈接

1.?創(chuàng)建表

1.1.?啟動(dòng)spark-sql

1.2.?建表參數(shù)

1.3.?創(chuàng)建非分區(qū)表

1.4.?創(chuàng)建分區(qū)表

1.5.?在已有的hudi表上創(chuàng)建新表

1.6.?通過CTAS (Create Table As Select)建表

2.?插入數(shù)據(jù)

2.1.?向非分區(qū)表插入數(shù)據(jù)

2.2.?向分區(qū)表動(dòng)態(tài)分區(qū)插入數(shù)據(jù)

2.3.?向分區(qū)表靜態(tài)分區(qū)插入數(shù)據(jù)

2.4.?使用bulk_insert插入數(shù)據(jù)

3.?查詢數(shù)據(jù)

3.1.?查詢

3.2.?時(shí)間旅行查詢

4.?更新數(shù)據(jù)

4.1.?update

4.2.?MergeInto

5.?刪除數(shù)據(jù)

6.?覆蓋數(shù)據(jù)

7.?修改表結(jié)構(gòu)（Alter Table）

8.?修改分區(qū)

9.?存儲(chǔ)過程（Procedures）

0. 相關(guān)文章鏈接

?Hudi文章匯總?

1.?創(chuàng)建表

1.1.?啟動(dòng)spark-sql

# 啟動(dòng)spark-sql之前需要先啟動(dòng)Hive的Metastore
nohup hive --service metastore & 

#針對(duì)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)境變量，手動(dòng)拷貝hive-site.xml到spark的conf下

1.2.?建表參數(shù)

1.3.?創(chuàng)建非分區(qū)表

• 創(chuàng)建一個(gè)cow表，默認(rèn)primaryKey 'uuid'，不提供preCombineField

create table hudi_cow_nonpcf_tbl (
  uuid int,
  name string,
  price double
) using hudi;

• 創(chuàng)建一個(gè)mor非分區(qū)表

create table hudi_mor_tbl (
  id int,
  name string,
  price double,
  ts bigint
) using hudi
tblproperties (
  type = 'mor',
  primaryKey = 'id',
  preCombineField = 'ts'
);

1.4.?創(chuàng)建分區(qū)表

創(chuàng)建一個(gè)cow分區(qū)外部表，指定primaryKey和preCombineField

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 '/tmp/hudi/hudi_cow_pt_tbl';

1.5.?在已有的hudi表上創(chuàng)建新表

不需要指定模式和非分區(qū)列（如果存在）之外的任何屬性，Hudi可以自動(dòng)識(shí)別模式和配置。

• 非分區(qū)表

create table hudi_existing_tbl0 
using hudi
location 'file:///tmp/hudi/dataframe_hudi_nonpt_table';

• 分區(qū)表

create table hudi_existing_tbl1 
using hudi
partitioned by (dt, hh)
location 'file:///tmp/hudi/dataframe_hudi_pt_table';

1.6.?通過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:///tmp/parquet_dataset/*.parquet';

# 通過CTAS加載數(shù)據(jù)
create table hudi_ctas_cow_pt_tbl2 
using hudi 
location 'file:/tmp/hudi/hudi_tbl/' 
options (
  type = 'cow',
  primaryKey = 'id',
  preCombineField = 'ts'
)
partitioned by (datestr) 
as 
select * from parquet_mngd
;

2.?插入數(shù)據(jù)

默認(rèn)情況下，如果提供了preCombineKey，則insert into的寫操作類型為upsert，否則使用insert。

2.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.2.?向分區(qū)表動(dòng)態(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;

2.3.?向分區(qū)表靜態(tài)分區(qū)插入數(shù)據(jù)

insert into hudi_cow_pt_tbl partition(dt = '2021-12-09', hh='11') select 2, 'a2', 1000;

2.4.?使用bulk_insert插入數(shù)據(jù)

hudi支持使用bulk_insert作為寫操作的類型，只需要設(shè)置兩個(gè)配置：

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

3.?查詢數(shù)據(jù)

3.1.?查詢

select fare, begin_lon, begin_lat, ts from  hudi_trips_snapshot where fare > 20.0

3.2.?時(shí)間旅行查詢

Hudi從0.9.0開始就支持時(shí)間旅行查詢。Spark SQL方式要求Spark版本 3.2及以上。

-- 關(guān)閉前面開啟的bulk_insert
set hoodie.sql.bulk.insert.enable=false;

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 '/tmp/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;

-- 基于第一次提交時(shí)間進(jìn)行時(shí)間旅行
select * from hudi_cow_pt_tbl1 timestamp as of '20220307091628793' where id = 1;

-- 其他時(shí)間格式的時(shí)間旅行寫法
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;

4.?更新數(shù)據(jù)

4.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';

4.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 ...])

• 執(zhí)行案例

-- 1、準(zhǔn)備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、準(zhǔn)備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)
;

5.?刪除數(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';

6.?覆蓋數(shù)據(jù)

• 使用INSERT_OVERWRITE類型的寫操作覆蓋分區(qū)表
• 使用INSERT_OVERWRITE_TABLE類型的寫操作插入覆蓋非分區(qū)表或分區(qū)表（動(dòng)態(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）通過動(dòng)態(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;

7.?修改表結(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');

8.?修改分區(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)表路徑的。刪除整個(gè)分區(qū)數(shù)據(jù)或直接刪除某個(gè)分區(qū)目錄并不精確。

9.?存儲(chǔ)過程（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)

• 案例（可用的存儲(chǔ)過程：Procedures | Apache Hudi）：

--show commit's info
call show_commits(table => 'hudi_cow_pt_tbl1', limit => 10);

注：其他Hudi相關(guān)文章鏈接由此進(jìn) ->??Hudi文章匯總?文章來源地址http://www.zghlxwxcb.cn/news/detail-457069.html

到了這里，關(guān)于Hudi（7）：Hudi集成Spark之spark-sql方式的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！