CDP集成Hudi实战-spark shell

[〇]关于本文

本文主要解释spark shell操作Hudi表的案例

软件版本
Hudi1.0.0
Hadoop Version3.1.1.7.3.1.0-197
Hive Version3.1.3000.7.3.1.0-197
Spark Version3.4.1.7.3.1.0-197
CDP7.3.1

[一]使用Spark-shell

1-配置hudi Jar包

[root@cdp73-1 ~]# for i in $(seq 1 6); do scp /opt/software/hudi-1.0.0/packaging/hudi-spark-bundle/target/hudi-spark3.4-bundle_2.12-1.0.0.jar   cdp73-$i:/opt/cloudera/parcels/CDH/lib/spark3/jars/; done
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 418.2MB/s   00:00
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 304.8MB/s   00:00
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 365.0MB/s   00:00
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 406.1MB/s   00:00
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 472.7MB/s   00:00
hudi-spark3.4-bundle_2.12-1.0.0.jar                                                                                                                                                                                          100%  105MB 447.1MB/s   00:00
[root@cdp73-1 ~]#

2-进入Spark-shell

spark-shell --packages org.apache.hudi:hudi-spark3.4-bundle_2.12:1.0.0 \
--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' \
--conf 'spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar'

3-初始化项目

// spark-shell
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.table.HoodieTableConfig._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
import org.apache.hudi.common.model.HoodieRecord
import spark.implicits._val tableName = "trips_table"
val basePath = "hdfs:///tmp/trips_table"

4-创建表

首次提交将自动初始化表,如果指定的基本路径中尚不存在该表。

5-导入数据

// spark-shell
val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),(1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),(1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),(1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo"    ),(1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"));var inserts = spark.createDataFrame(data).toDF(columns:_*)
inserts.write.format("hudi").option("hoodie.datasource.write.partitionpath.field", "city").option("hoodie.embed.timeline.server", "false").option("hoodie.table.name", tableName).mode(Overwrite).save(basePath)

【映射到Hudi写操作】​​​​​​​Hudi提供了多种写操作——包括批量和增量写操作——以将数据写入Hudi表,这些操作具有不同的语义和性能。当未配置记录键(请参见下面的键)时,将选择bulk_insert作为写操作,这与Spark的Parquet数据源的非默认行为相匹配。

6-查询数据

// spark-shell
val tripsDF = spark.read.format("hudi").load(basePath)
tripsDF.createOrReplaceTempView("trips_table")spark.sql("SELECT uuid, fare, ts, rider, driver, city FROM  trips_table WHERE fare > 20.0").show()
spark.sql("SELECT _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare FROM  trips_table").show()

7-更新数据

// Lets read data from target Hudi table, modify fare column for rider-D and update it. 
val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-D").withColumn("fare", col("fare") * 10)updatesDf.write.format("hudi").option("hoodie.datasource.write.operation", "upsert").option("hoodie.embed.timeline.server", "false").option("hoodie.datasource.write.partitionpath.field", "city").option("hoodie.table.name", tableName).mode(Append).save(basePath)

8-合并数据

// spark-shell
val adjustedFareDF = spark.read.format("hudi").load(basePath).limit(2).withColumn("fare", col("fare") * 10)
adjustedFareDF.write.format("hudi").
option("hoodie.embed.timeline.server", "false").
mode(Append).
save(basePath)
// Notice Fare column has been updated but all other columns remain intact.
spark.read.format("hudi").load(basePath).show()

9-删除数据

/ spark-shell
// Lets  delete rider: rider-D
val deletesDF = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-F")deletesDF.write.format("hudi").option("hoodie.datasource.write.operation", "delete").option("hoodie.datasource.write.partitionpath.field", "city").option("hoodie.table.name", tableName).option("hoodie.embed.timeline.server", "false").mode(Append).save(basePath)

​​​​​​​

10-数据索引

import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.table.HoodieTableConfig._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
import org.apache.hudi.common.model.HoodieRecord
import spark.implicits._val tableName = "trips_table_index"
val basePath = "hdfs:///tmp/hudi_indexed_table"val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),(1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),(1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),(1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo"    ),(1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"));var inserts = spark.createDataFrame(data).toDF(columns:_*)
inserts.write.format("hudi").option("hoodie.datasource.write.partitionpath.field", "city").option("hoodie.table.name", tableName).option("hoodie.write.record.merge.mode", "COMMIT_TIME_ORDERING").option(RECORDKEY_FIELD_OPT_KEY, "uuid").option("hoodie.embed.timeline.server", "false").mode(Overwrite).save(basePath)// Create record index and secondary index for the table
spark.sql(s"CREATE TABLE hudi_indexed_table USING hudi LOCATION '$basePath'")
// Create bloom filter expression index on driver column
spark.sql(s"CREATE INDEX idx_bloom_driver ON hudi_indexed_table USING bloom_filters(driver) OPTIONS(expr='identity')");
// It would show bloom filter expression index 
spark.sql(s"SHOW INDEXES FROM hudi_indexed_table");
// Query on driver column would prune the data using the idx_bloom_driver index
spark.sql(s"SELECT uuid, rider FROM hudi_indexed_table WHERE driver = 'driver-S'");// Create column stat expression index on ts column
spark.sql(s"CREATE INDEX idx_column_ts ON hudi_indexed_table USING column_stats(ts) OPTIONS(expr='from_unixtime', format = 'yyyy-MM-dd')");
// Shows both expression indexes 
spark.sql(s"SHOW INDEXES FROM hudi_indexed_table");
// Query on ts column would prune the data using the idx_column_ts index
spark.sql(s"SELECT * FROM hudi_indexed_table WHERE from_unixtime(ts, 'yyyy-MM-dd') = '2023-09-24'");// To create secondary index, first create the record index
spark.sql(s"SET hoodie.metadata.record.index.enable=true");
spark.sql(s"CREATE INDEX record_index ON hudi_indexed_table (uuid)");
// Create secondary index on rider column
spark.sql(s"CREATE INDEX idx_rider ON hudi_indexed_table (rider)");// Expression index and secondary index should show up
spark.sql(s"SHOW INDEXES FROM hudi_indexed_table");
// Query on rider column would leverage the secondary index idx_rider
spark.sql(s"SELECT * FROM hudi_indexed_table WHERE rider = 'rider-E'");// Update a record and query the table based on indexed columns
spark.sql(s"UPDATE hudi_indexed_table SET rider = 'rider-B', driver = 'driver-N', ts = '1697516137' WHERE rider = 'rider-A'");
// Data skipping would be performed using column stat expression index
spark.sql(s"SELECT uuid, rider FROM hudi_indexed_table WHERE from_unixtime(ts, 'yyyy-MM-dd') = '2023-10-17'");
// Data skipping would be performed using bloom filter expression index
spark.sql(s"SELECT * FROM hudi_indexed_table WHERE driver = 'driver-N'");
// Data skipping would be performed using secondary index
spark.sql(s"SELECT * FROM hudi_indexed_table WHERE rider = 'rider-B'");// Drop all the indexes
spark.sql(s"DROP INDEX secondary_index_idx_rider on hudi_indexed_table");
spark.sql(s"DROP INDEX record_index on hudi_indexed_table");
spark.sql(s"DROP INDEX expr_index_idx_bloom_driver on hudi_indexed_table");
spark.sql(s"DROP INDEX expr_index_idx_column_ts on hudi_indexed_table");
// No indexes should show up for the table
spark.sql(s"SHOW INDEXES FROM hudi_indexed_table");spark.sql(s"SET hoodie.metadata.record.index.enable=false");

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/66371.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Python爬虫基础——百度新闻页面结构剖析

经过上一篇文章文章[Python爬虫基础——认识网页结构(各种标签的使用)]的介绍,我们对网页结构已经有了初步的认识,本篇文章针对百度新闻界界面结构进行剖析。 在浏览器地址栏中输入https://news.baidu.com/,然后按住F12打开发这工具在“Eleme…

【老白学 Java】保存 / 恢复对象状态

保存、恢复对象状态 文章来源:《Head First Java》修炼感悟。 上两篇文章分别讨论了对象序列化和反序列化,主要是针对数据文件进行读、写操作的介绍。 本篇文章通过一个完整的例子,复习一下对象保存与恢复的操作步骤,在文章最后做…

进程间通信——网络通信——UDP

进程间通信(分类):网络通信、无名管道、有名管道、信号、消息队列、共享内存、信号量集 OSI七层模型:(理论模型) 应用层 : 要传输的数据信息,如文件传输,电子邮件等 表示层 : 数…

3272 小蓝的漆房

将devc设置支持编译就能用新的遍历方式 for(auto &x : s)//遍历容器s,变量为x /* 多循环的嵌套: 计数是否需要重置为0; 是否因为ans定义成全局变量导致ans在比较多时候会出现错误*/ /* 1.对于一个标准色,对目标数组遍历, 如…

海外云服务器能用来做什么?

海外云服务器不仅服务种类繁多,而且能满足多行业的需求,方便了越来越多的企业与个人。本文将探讨海外云服务器的核心服务及其适用领域,帮助企业更好地了解这一技术资源。 云存储:安全高效的数据管理 海外云服务器为用户提供了稳定…

导出中心设计

业务背景 应用业务经常需要导出数据,但是并发的导出以及不合理的导出参数常常导致应用服务的内存溢出、其他依赖应用的崩溃、导出失败;因此才有导出中心的设计 设计思想 将导出应用所需的内存转移至导出中心,将导出的条数加以限制&#xf…

智能工厂的设计软件 应用场景的一个例子: 为AI聊天工具添加一个知识系统 之23 “单子”职业能力原型:PIN语言/AI操作系统/robot扮演的actor

本文提要 很重要的一点是,PIN语言的每个 “词项”(最小表达单子) 唯一地提供本项目模板的一个“ 槽”位--占位符变量。这样确定了 本项目的“祖传代码” 的脚本模板了 我前面已经为三套文 给出了 对应的三套模板的名称和用意了--”在本项目的…

Re77 读论文:LoRA: Low-Rank Adaptation of Large Language Models

诸神缄默不语-个人CSDN博文目录 诸神缄默不语的论文阅读笔记和分类 论文全名:LoRA: Low-Rank Adaptation of Large Language Models ArXiv网址:https://arxiv.org/abs/2106.09685 官方GitHub网站(包含在RoBERTa、DeBERTa、GPT-2上用Lora微调…

Vue3苦逼的学习之路

从一名测试转战到全栈是否可以自学做到,很多朋友肯定会说不可能,或就算转了也是个一般水平,我很认同,毕竟没有经过各种项目的摧残,但是还是得踏足一下这个领域。所以今天和大家分享vue3中的相关内容,大佬勿…

C++单例模式跨DLL调用问题梳理

问题案例: 假设有这样一个单例模式的代码 //test.h header class Test { public:static Test &instance() {static Test ins;return ins;}void foo(); };void testFoo();//test.cpp source #include "test.h"void Test::foo() {printf("%p\n&q…

ESP32-S3系统级芯片支持烧录的编程语言

ESP32-S3作为一款功能强大的MCU系统级芯片,支持多种编程语言的烧录和开发。以下是对ESP32-S3支持的主要编程语言的详细介绍: 一、C/C ESP-IDF框架:ESP32-S3支持使用乐鑫官方的ESP-IDF(Espressif IoT Development Framework&…

Redis 数据库源码分析

Redis 数据库源码分析 我们都知道Redis是一个 <key,value> 的键值数据库&#xff0c;其实也就是一个 Map。如果让我来实现这样一个 Map&#xff0c;我肯定是用数组&#xff0c;当一个 key 来的时候&#xff0c;首先进行 hash 运算&#xff0c;接着对数据的 length 取余&…

我的nvim的init.lua配置

nvim的配置文件路径在&#xff5e;/.config/nvim路径下&#xff1a; 一、目录如下&#xff1a; coc-settings.json文件是配置代码片段路径的文件init.lua配置文件的启动脚本lua/config.lua 全局配置文件lua/keymaps.lua 快捷键映射键文件lua/plugins.lua 插件的安装和配置文件…

权限掩码umask

1 、 设置新建文件或目录的默认权限 在 Linux 系统中&#xff0c;当用户创建一个新的文件或目录时&#xff0c;系统都会为新建的文件或目录分配默认的权限&#xff0c;该默认权限与umask 值有关&#xff0c;其具体关系是&#xff1a; 新建文件的默认权限 0666-umask 值 新建…

Kubernetes Gateway API-5-后端协议和网关基础设置标签

1 后端协议 自 v1.2.0 开始支持 并非所有网关API实现都支持自动协议选择。在某些情况下&#xff0c;协议在没有明确选择加入的情况下被禁用。 当 Route 的后端引用Kubernetes Service 时&#xff0c;应用程序开发人员可以使用 ServicePort appProtocol 字段指定协议。 例如…

C++语言的网络编程

C语言的网络编程 引言 随着互联网的迅猛发展&#xff0c;网络编程已成为软件开发的重要组成部分。C作为一种高效的编程语言&#xff0c;因其出色的性能和灵活性&#xff0c;广泛应用于网络编程领域。本文将介绍C网络编程的基本概念、常用的网络库&#xff0c;以及一些具体的应…

考试座位号(PTA)C语言

每个 PAT 考生在参加考试时都会被分配两个座位号&#xff0c;一个是试机座位&#xff0c;一个是考试座位。正常情况下&#xff0c;考生在入场时先得到试机座位号码&#xff0c;入座进入试机状态后&#xff0c;系统会显示该考生的考试座位号码&#xff0c;考试时考生需要换到考试…

宝安湾区之光附近的钓鱼点

工作日的午休我经常在公司附近骑行&#xff0c;有时候也会骑行到宝安的湾区之光。但是我最感兴趣的除了湾区之光摩天轮&#xff0c;还有雷打不动的快乐钓鱼佬。 上图红框区域的河岸每天都会出现零零散散的快乐钓鱼佬&#xff0c;他们好像都有自己的钓鱼窝点。我发现来这里钓鱼也…

GNU链接器简介-3

GNU链接器简介-3 1 SECTIONS Command1.1 Output Section Description2.2 Output Section Name1.3 Output Section Address1.4 Input Section Description1.4.1 Input Section Basics1.4.2 Input Section Wildcard Patterns1.4.3 Input Section for Common Symbols1.4.4 Input S…

【练习】PAT 乙 1022 D进制的A+B

题目 输入两个非负10进制整数A和B(<2^30-1)&#xff0c;输出AB的D (1 < D < 10)进制数。 输入格式 输入在一行中依次给出3个整数A、B和D。 输出格式 输出AB的D进制数。 输入样例 123 456 8 输出样例 1103 来源&#xff1a;PAT 乙 1022 D进制的AB ——————————…