使用Spark操作Hudi表详细教程_spark sql操作hudi
mvn clean package -Dflink1.15 -Dscala2.12 -Dspark3.3 -DskipTests -Pflink-bundle-shade-hive3 -T 4
编译输出的Spark Hudi依赖位于hudi/packaging/hudi-spark-bundle/target，将其中的hudi-spark3.x-bundle_2.12-0.xx.x.jar复制走备用。
环境配置
需要禁用Yarn组件的yarn.timeline-service.enabled配置。修改完毕后重启Yarn组件。
或者是在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;
验证：
select \* from hudi_mor_tbl;
发现只有新增的这一条数据。
删除数据
SQL方式：
delete from hudi_mor_tbl where id % 2 = 1;
验证：
select \* from hudi_mor_tbl;
Kerberos和权限配置
例如，如果要允许Hudi用户对Hudi表进行操作，提交队列为default，表路径为hdfs:///hudi/t1，可以通过以下步骤使用Ranger进行设置：
1、在Ranger中创建一个名为hudi的用户。
2、分配给hudi用户以下目录的读写权限：/hdfs/hudi/t1，/tmp，/user/hudi。
3、赋予hudi用户对yarn default队列的权限。
自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。
深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！
因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！
由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新
如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
-1712880612940)]
[外链图片转存中…(img-AWu5eSZ1-1712880612940)]
[外链图片转存中…(img-BUqtq0q6-1712880612941)]
[外链图片转存中…(img-oQ4yreuf-1712880612942)]
[外链图片转存中…(img-CtjFCdRv-1712880612942)]
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！
由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新
如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-FXBtP8pK-1712880612943)]
腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。
更多推荐
19
0- 0
已为社区贡献4条内容
所有评论(0)