目录1. 摘要2. 环境准备2.1 启动spark-sql2.2 设置并发度3. Create Table4. Insert Into4.1 Insert4.2 Select5. Update5.1 Update5.2 Select6. Delete6.1 Delete6.2 Select7. Merge Into7.1 Merge Into Insert7.2 Select7.4 Merge Into Update7.5 Select7.6 Merge Into Delete8. 删除表9. 总结
社区小伙伴一直期待的Hudi整合Spark SQL的PR正在积极Review中并已经快接近尾声，Hudi集成Spark SQL预计会在下个版本正式发布，在集成Spark SQL后，会极大方便用户对Hudi表的DDL/DML操作，下面就来看看如何使用Spark SQL操作Hudi表。
首先需要将PR拉取到本地打包，生成SPARK_BUNDLE_JAR(hudi-spark-bundle_2.11-0.9.0-SNAPSHOT.jar)包
在配置完spark环境后可通过如下命令启动spark-sql
spark-sql –jars $PATH_TO_SPARK_BUNDLE_JAR –conf ‘spark.serializer=org.apache.spark.serializer.KryoSerializer’ –conf ‘spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension’
由于Hudi默认upsert/insert/delete的并发度是1500，对于演示的小规模数据集可设置更小的并发度。
set hoodie.upsert.shuffle.parallelism=1;
set hoodie.insert.shuffle.parallelism=1;
set hoodie.delete.shuffle.parallelism=1;
set hoodie.insert.shuffle.parallelism=1;
set hoodie.delete.shuffle.parallelism=1;
同时设置不同步Hudi表元数据
set hoodie.datasource.meta.sync.enable=false;
使用如下SQL创建表
create table test_hudi_table (
id int,
name string,
price double,
ts long,
dt string
) using hudi
partitioned by (dt)
options (
primaryKey=’id’,
type=’mor’
)
location ‘file:///tmp/test_hudi_table’
id int,
name string,
price double,
ts long,
dt string
) using hudi
partitioned by (dt)
options (
primaryKey=’id’,
type=’mor’
)
location ‘file:///tmp/test_hudi_table’
说明：表类型为MOR，主键为id，分区字段为dt，合并字段默认为ts。
创建Hudi表后查看创建的Hudi表
show create table test_hudi_table
使用如下SQL插入一条记录
insert into test_hudi_table select 1 as id, ‘hudi’ as name, 10 as price, 1000 as ts, ‘2021-05-05’ as dt
insert完成后查看Hudi表本地目录结构，生成的元数据、分区和数据与Spark Datasource写入均相同。
使用如下SQL查询Hudi表数据
select * from test_hudi_table
查询结果如下
使用如下SQL将id为1的price字段值变更为20
update test_hudi_table set price=20.0 where id=1
再次查询Hudi表数据
select * from test_hudi_table
查询结果如下，可以看到price已经变成了20.0
查看Hudi表的本地目录结构如下，可以看到在update之后又生成了一个，同时生成了一个增量log文件。
使用如下SQL将id=1的记录删除
delete from test_hudi_table where id=1
查看Hudi表的本地目录结构如下，可以看到delete之后又生成了一个，同时生成了一个增量log文件。
再次查询Hudi表
select * from test_hudi_table;
查询结果如下，可以看到已经查询不到任何数据了，表明Hudi表中已经不存在任何记录了。
使用如下SQL向插入数据
merge into test_hudi_table as t0
using (
select 1 as id, ‘a1’ as name, 10 as price, 1000 as ts, ‘2021-03-21’ as dt
) as s0
on t0.id=s0.id
when not matched and s0.id % 2=1 then insert *
using (
select 1 as id, ‘a1’ as name, 10 as price, 1000 as ts, ‘2021-03-21’ as dt
) as s0
on t0.id=s0.id
when not matched and s0.id % 2=1 then insert *
查询Hudi表数据
select * from test_hudi_table
查询结果如下，可以看到Hudi表中存在一条记录
使用如下SQL更新数据
merge into test_hudi_table as t0
using (
select 1 as id, ‘a1’ as name, 12 as price, 1001 as ts, ‘2021-03-21’ as dt
) as s0
on t0.id=s0.id
when matched and s0.id % 2=1 then update set *
using (
select 1 as id, ‘a1’ as name, 12 as price, 1001 as ts, ‘2021-03-21’ as dt
) as s0
on t0.id=s0.id
when matched and s0.id % 2=1 then update set *
查询Hudi表
select * from test_hudi_table
查询结果如下，可以看到Hudi表中的分区已经更新了
使用如下SQL删除数据
merge into test_hudi_table t0
using (
select 1 as s_id, ‘a2’ as s_name, 15 as s_price, 1001 as s_ts, ‘2021-03-21’ as dt
) s0
on t0.id=s0.s_id
when matched and s_ts=1001 then delete
using (
select 1 as s_id, ‘a2’ as s_name, 15 as s_price, 1001 as s_ts, ‘2021-03-21’ as dt
) s0
on t0.id=s0.s_id
when matched and s_ts=1001 then delete
查询结果如下，可以看到Hudi表中已经没有数据了
使用如下命令删除Hudi表
drop table test_hudi_table;
使用show tables查看表是否存在
show tables;
可以看到已经没有表了
通过上面示例简单展示了通过Spark SQL Insert/Update/Delete Hudi表数据，通过SQL方式可以非常方便地操作Hudi表，降低了使用Hudi的门槛。另外Hudi集成Spark SQL工作将继续完善语法，尽量对标Snowflake和BigQuery的语法，如插入多张表（INSERT ALL WHEN condition1 INTO t1 WHEN condition2 into t2），变更Schema以及CALL Cleaner、CALL Clustering等Hudi表服务。
以上就是Apache Hudi集成Spark SQL操作hide表的详细内容，更多关于Apache Hudi集成Spark SQL的资料请关注脚本之家其它相关文章！
您可能感兴趣的文章:详解Spark?Sql在UDF中如何引用外部数据大数据Spark Sql中日期转换FROM_UNIXTIME和UNIX_TIMESTAMP的使用springboot集成spark并使用spark-sql的示例详解教你如何让spark?sql写mysql的时候支持update操作Spark?SQL配置及使用教程Spark SQL 编程初级实践详解
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
