使用Spark操作Hudi表详细教程_spark读取hudi，2024年最新大数据开发开发必须要会
执行作业前建议导入如下：插入数据验证：普通查询增量查询首先再插入/修改一条数据，参见插入/修改数据。然后执行：发现只取出了最近插入/修改后的数据。验证方法使用普通查询。验证方法使用普通查询。发现只有新增的这一条数据。验证方法使用普通查询。启动Hudi spark sql的方法：如果使用Hudi的版本为0.11.x，需要执行：创建表：验证：插入数据SQL方式：验证：普通查询SQL方式：修改数据SQL
执行作业前建议导入如下：
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和权限配置
自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。
深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！
因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！
由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新
如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
gjasu-1713012878195)]
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！
由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新
如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-pHdfYi09-1713012878195)]
一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
永洪科技，致力于打造全球领先的数据技术厂商，具备从数据应用方案咨询、BI、AIGC智能分析、数字孪生、数据资产、数据治理、数据实施的端到端大数据价值服务能力。
更多推荐
6
0- 0
已为社区贡献3条内容
所有评论(0)