使用Spark操作Hudi表详细教程_spark读取hudi

【代码】使用Spark操作Hudi表详细教程_spark读取hudi。

2401_84184689

735人浏览 · 2024-04-12 08:11:56

2401_84184689 · 2024-04-12 08:11:56 发布


编译输出的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()


发现只取出了最近插入/修改后的数据。


### 修改数据

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

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”)


验证方法使用普通查询。发现只有新增的这一条数据。


### 删除数据

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队列的权限。


如果启用了Kerberos，还需要执行以下额外步骤：


**自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。**

**深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！**

**因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。**
![img](https://img-blog.csdnimg.cn/img_convert/eb6b1f8651a7f9b9707a89a3d4d56962.png)
![img](https://img-blog.csdnimg.cn/img_convert/3e2bff78f8f7a4cddb4a722230d95c5f.png)
![img](https://img-blog.csdnimg.cn/img_convert/625b3f29aa70fb845433c93322b040a4.png)
![img](https://img-blog.csdnimg.cn/img_convert/eae5383a8e7939921d382349536eb442.png)
![img](https://img-blog.csdnimg.cn/img_convert/404dbf4686ab4ea8b17af1e4cd913495.png)

**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！**

**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**

**如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）**
![img](https://img-blog.csdnimg.cn/img_convert/bd4e89ce2d4af595d9e8cead67536179.png)

-1712880647364)]
[外链图片转存中...(img-kXIORuTp-1712880647364)]
[外链图片转存中...(img-7tE6xyMk-1712880647365)]
[外链图片转存中...(img-WvJhsYBp-1712880647365)]
[外链图片转存中...(img-SxpP6V8n-1712880647366)]

**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！**

**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**

**如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）**
[外链图片转存中...(img-ujLe5kE5-1712880647366)]