2024年使用Spark操作Hudi表详细教程_spark sql操作hudi(1)，2024年最新大数据开发事件分发机制收藏这一篇就够了
或者是在spark-defaults.conf中增加spark.hadoop.yarn.timeline-service.enabled=false。建议这样配置，避免修改Yarn的全局配置。接着将Hudi编译之后的hudi-spark3.x-bundle_2.12-0.xx.x.jar复制到${SPARK_HOME}/jars目录中。
·
网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。
一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
或者是在spark-defaults.conf中增加spark.hadoop.yarn.timeline-service.enabled=false。建议这样配置，避免修改Yarn的全局配置。
接着将Hudi编译之后的hudi-spark3.x-bundle_2.12-0.xx.x.jar复制到${SPARK_HOME}/jars目录中。
Spark Shell方式
启动Hudi spark shell的方法：
./spark-shell \
--master yarn \
--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'
如果使用Hudi的版本为0.11.x，需要执行：
./spark-shell \
--master yarn \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer'
执行作业前建议导入如下：
import org.apache.hudi.QuickstartUtils._
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._
插入数据
import org.apache.spark.sql._
import org.apache.spark.sql.types._
val fields = Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("price", DoubleType, true),
StructField("ts", LongType, true)
)
val simpleSchema = StructType(fields)
val data = Seq(Row(2, "a2", 200.0, 100L))
val df = spark.createDataFrame(data, simpleSchema)
df.write.format("hudi").
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "id").
option(TABLE_NAME, "hudi\_mor\_tbl\_shell").
option(TABLE_TYPE_OPT_KEY, "MERGE\_ON\_READ").
mode(Append).
save("hdfs:///hudi/hudi\_mor\_tbl\_shell")
验证：
val df = spark.
read.
format("hudi").
load("hdfs:///hudi/hudi\_mor\_tbl\_shell")
df.createOrReplaceTempView("hudi\_mor\_tbl\_shell")
spark.sql("select \* from hudi\_mor\_tbl\_shell").show()
普通查询
val df = spark.
read.
format("hudi").
load("hdfs:///hudi/hudi\_mor\_tbl\_shell")
df.createOrReplaceTempView("hudi\_mor\_tbl\_shell")
spark.sql("select \* from hudi\_mor\_tbl\_shell").show()
增量查询
首先再插入/修改一条数据，参见插入/修改数据。然后执行：
spark.
read.
format("hudi").
load("hdfs:///hudi/hudi\_mor\_tbl\_shell").
createOrReplaceTempView("hudi\_mor\_tbl\_shell")
val commits = spark.sql("select distinct(\_hoodie\_commit\_time) as commitTime from hudi\_mor\_tbl\_shell order by commitTime desc").map(k => k.getString(0)).take(50)
val beginTime = commits(commits.length - 1)
val idf = spark.read.format("hudi").
option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL).
option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).
load("hdfs:///hudi/hudi\_mor\_tbl\_shell")
idf.createOrReplaceTempView("hudi\_mor\_tbl\_shell\_incremental")
spark.sql("select `\_hoodie\_commit\_time`, id, name, price, ts from hudi\_mor\_tbl\_shell\_incremental").show()
发现只取出了最近插入/修改后的数据。
修改数据
import org.apache.spark.sql._
import org.apache.spark.sql.types._
val fields = Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("price", DoubleType, true),
StructField("ts", LongType, true)
)
val simpleSchema = StructType(fields)
val data = Seq(Row(2, "a2", 400.0, 2222L))
val df = spark.createDataFrame(data, simpleSchema)
df.write.format("hudi").
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "id").
option(TABLE_NAME, "hudi\_mor\_tbl\_shell").
option(TABLE_TYPE_OPT_KEY, "MERGE\_ON\_READ").
mode(Append).
save("hdfs:///hudi/hudi\_mor\_tbl\_shell")
验证方法使用普通查询。
Insert overwrite
import org.apache.spark.sql._
import org.apache.spark.sql.types._
val fields = Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("price", DoubleType, true),
StructField("ts", LongType, true)
)
val simpleSchema = StructType(fields)
val data = Seq(Row(99, "a99", 20.0, 900L))
val df = spark.createDataFrame(data, simpleSchema)
df.write.format("hudi").
option(OPERATION.key(),"insert\_overwrite").
option(PRECOMBINE_FIELD.key(), "ts").
option(RECORDKEY_FIELD.key(), "id").
option(TBL_NAME.key(), "hudi\_mor\_tbl\_shell").
option(TABLE_TYPE_OPT_KEY, "MERGE\_ON\_READ").
mode(Append).
save("hdfs:///hudi/hudi\_mor\_tbl\_shell")
验证方法使用普通查询。发现只有新增的这一条数据。
删除数据
import org.apache.spark.sql._
import org.apache.spark.sql.types._
val fields = Array(
StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("price", DoubleType, true),
StructField("ts", LongType, true)
)
val simpleSchema = StructType(fields)
val data = Seq(Row(2, "a2", 400.0, 2222L))
val df = spark.createDataFrame(data, simpleSchema)
df.write.format("hudi").
option(OPERATION_OPT_KEY,"delete").
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "id").
option(TABLE_NAME, "hudi\_mor\_tbl\_shell").
mode(Append).
save("hdfs:///hudi/hudi\_mor\_tbl\_shell")
验证方法使用普通查询。
Spark SQL方式
启动Hudi spark sql的方法：
./spark-sql \
--master yarn \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--conf 'spark.sql.catalog.spark\_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'
如果使用Hudi的版本为0.11.x，需要执行：
./spark-sql \
--master yarn \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'
创建表：
create table hudi_mor_tbl (
id int,
name string,
price double,
ts bigint
) using hudi
tblproperties (
type = 'mor',
primaryKey = 'id',
preCombineField = 'ts'
)
location 'hdfs:///hudi/hudi\_mor\_tbl';
验证：
show tables;
插入数据
SQL方式：
insert into hudi_mor_tbl select 1, 'a1', 20, 1000;
验证：
select \* from hudi_mor_tbl;
普通查询
SQL方式：
select \* from hudi_mor_tbl;
修改数据
SQL方式：
update hudi_mor_tbl set price = price \* 2, ts = 1111 where id = 1;
验证：
select \* from hudi_mor_tbl;
insert overwrite
SQL方式：
insert overwrite hudi_mor_tbl select 99, 'a99', 20.0, 900;
网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。
一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
_tbl select 99, ‘a99’, 20.0, 900;
[外链图片转存中...(img-i2Yf1FmP-1714857839539)]
[外链图片转存中...(img-Qt40xiHu-1714857839539)]
**网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**
**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**
**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**
魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。
更多推荐
5
0- 0
已为社区贡献2条内容
所有评论(0)