Iceberg与SparkSQL写操作整合

devtools/2024/9/23 3:47:59/

前言

spark操作iceberg之前先要配置spark catalogs，详情参考Iceberg与Spark整合环境配置。

有些操作需要在spark3中开启iceberg sql扩展。

Iceberg使用Apache Spark的DataSourceV2 API来实现数据源和catalog。Spark DSv2是一个不断发展的API，在Spark版本中具有不同级别的支持:
在这里插入图片描述
Spark 3支持SQL INSERT INTO、MERGE INTO和INSERT OVERWRITE，以及新的DataFrameWriterV2 API来进行iceberg表的写操作，接下来我们进行详细讲解。

INSERT INTO

insert into是往iceberg表中插入新数据，主要有两种语法：

INSERT INTO prod.db.table VALUES (1, 'a'), (2, 'b')

INSERT INTO prod.db.table SELECT ...

这两种语法和其它组件如hive等没有太多区别，比较容易掌握。

MERGE INTO

Iceberg "merge into"语法可以对表数据进行行级更新或删除，在Spark3.x版本之后支持，其原理是重写包含需要删除和更新行数据所在的data files。"merge into"可以使用一个查询结果数据来更新目标表的数据，其语法通过类似join关联方式，根据指定的匹配条件对匹配的行数据进行相应操作。

  1. 语法
MERGE INTO tbl t -- 目标表
USING (SELECT ...) s -- 数据源表，也就是用数据源表查出的数据来更新或删除目标表
ON t.id = s.id  -- 关联条件，类似join的on条件
WHEN MATCHED AND ... THEN DELETE -- 删除直接用delete命令
WHEN MATCHED AND ... THEN UPDATE SET ... --更新用upate set
WHEN MATCHED AND ... AND ... THEN UPDATE SET ... --多条件更新
WHEN NOT MATCHED ADN ... THEN INSERT (col1,col2...) VALUES(s.col1,s.col2 ...) --匹配不上向目标表插入数据
  1. 示例
  • 创建两张表a和b
create table  hadoop_prod.default.a (id int,name string,age int) using iceberg;

create table  hadoop_prod.default.b (id int,name string,age int,tp string) using iceberg
  • 插入数据
insert into hadoop_prod.default.a values (1,"zs",18),(2,"ls",19),(3,"ww",20)

insert into hadoop_prod.default.b values (1,"zs",30,"delete"),(2,"李四",31,"update"),(4,"王五",32,"add")
  • 使用MERGE INTO 语法向目标表更新、删除、新增数据
    这里我们计划将b表与a表匹配id，如果b表中tp字段是"delete"那么a表中对应的id数据删除，如果b表中tp字段是"update"，那么a表中对应的id数据其他字段进行更新，如果a表与b表id匹配不上，那么将b表中的数据插入到a表中，具体操作如下：
merge into hadoop_prod.default.a  t1  -- 目标表a
using (select id,name ,age,tp from hadoop_prod.default.b) t2 -- 数据源表b
on t1.id = t2.id -- 关联条件为id
when matched and t2.tp = 'delete' then delete -- 如果数据源表中tp字段为delete，则对目标表关联d对应的数据进行删除操作
when matched and t2.tp = 'update' then update set t1.name = t2.name,t1.age = t2.age -- 如果数据源表tp字段为update，则对目标表关联id对应数据用数据源表中name和age更新目标表对应字段
when not matched then insert (id,name,age) values (t2.id,t2.name,t2.age) -- 如果id关联不上，则直接把数据源表对应id这条数据插入到目标表中

注意：我们很多数据库都没有类似merge into的操作，为了便于初学者理解，每一行操作都有详细的注释。

  • 结果
    在这里插入图片描述
    id=1，可以匹配上，但数据源表tp为delete，因此会把目标表id=1对应的行删除；
    id=2，可以匹配上，但数据源表tp为update，因此会把目标表id=2对应的name和age用数据源表name和age进行更新；
    id=3，没有匹配上，需要把数据源表对应的这条数据插入到目标表，但是由于数据源中没有id=3的数据，因此没有插入数据，此时保留数据源表中id=3对应的数据；
    id=4，没有匹配上，需要把数据源表对应的这条数据插入到目标表；

注意：更新数据时，在查询的数据中只能有一条匹配的数据更新到目标表，否则将报错。

INSERT OVERWRITE

"insert overwrite"可以覆盖Iceberg表中的数据，这种操作会将表中全部数据替换掉，建议如果有部分数据替换操作可以使用"merge into"操作。

对于Iceberg分区表使用"insert overwrite"操作时，有两种情况，第一种是“动态覆盖”，第二种是“静态覆盖”。

  1. 动态分区覆盖
    动态覆盖会全量将原有数据覆盖，并将新插入的数据根据Iceberg表分区规则自动分区，类似Hive中的动态分区。

  2. 静态分区覆盖
    静态覆盖需要在向Iceberg中插入数据时需要手动指定分区，如果当前Iceberg表存在这个分区，那么只有这个分区的数据会被覆盖，其他分区数据不受影响，如果Iceberg表不存在这个分区，那么相当于给Iceberg表增加了个一个分区。

  3. 示例

  • 创建三张表并插入数据
    创建test1分区表、test2普通表、test3普通表三张表，并插入数据，每张表字段相同，但是插入数据不同。
-- test1为分区表
create table  hadoop_prod.default.test1 (id int,name string,loc string)
using iceberg
partitioned by (loc);-- 插入数据
insert into hadoop_prod.default.test1 values (1,"zs","beijing"),(2,"ls","shanghai");

-- test2为普通无分区表
create table  hadoop_prod.default.test2 (id int,name string,loc string)
using iceberg;
-- 插入数据
insert into hadoop_prod.default.test2 values (10,"x1","shandong"),(11,"x2","hunan");

-- test3为普通无分区表
create table  hadoop_prod.default.test3 (id int,name string,loc string)
using iceberg;
-- 插入数据
insert into hadoop_prod.default.test3 values (3,"ww","beijing"),(4,"ml","shanghai"),(5,"tq","guangzhou");
  • 使用insert overwrite 读取test3表中的数据覆盖到test2表中
-- 使用insert overwrite 读取test3 表中的数据覆盖到test2 普通表中
insert overwrite hadoop_prod.default.test2 select id,name,loc from  hadoop_prod.default.test3;
-- 查询test2表数据
select * from hadoop_prod.default.test2;

此时test2表中的结果如下：
在这里插入图片描述
说明此时insert overwrite操作是把test2表的数据全部删除，然后把test3表的所有数据插入到test2表。

  • 使用insert overwrite 读取test3表数据，动态分区方式覆盖到表test1
-- 使用insert overwrite 读取test3表数据 动态分区方式覆盖到表 test1
insert overwrite hadoop_prod.default.test1 select id,name,loc from  hadoop_prod.default.test3;
-- 查询 test1 表数据
select * from hadoop_prod.default.test1;

此时test1表中的数据如下：
在这里插入图片描述
说明此时insert overwrite操作是把test1表的数据全部删除，然后把test3表的所有数据插入到test1表，并且分区字段loc按照动态分区的方式进行分区。

  • 静态分区方式，将iceberg表test3的数据覆盖到Iceberg表test1中
    这里可以将test1表删除，然后重新创建，加载数据，也可以直接读取test3中的数据静态分区方式更新到test1。另外，使用insert overwrite 语法覆盖静态分区方式时，查询的语句中就不要再次写入分区列，否则会重复。
-- 删除表test1,重新创建表test1 分区表，并插入数据
drop table hadoop_prod.default.test1;
-- 重建test1分区表
create table  hadoop_prod.default.test1 (id int,name string,loc string) using iceberg partitioned by (loc);
-- 插入数据
insert into hadoop_prod.default.test1 values (1,"zs","beijing"),(2,"ls","shanghai");
-- 查询test1表数据
select * from hadoop_prod.default.test1;

在这里插入图片描述

-- 注意：指定静态分区"jiangsu",静态分区下，就不要在查询 “loc" 列了，否则重复
insert overwrite hadoop_prod.default.test1 partition (loc = "jiangsu") select id,name from  hadoop_prod.default.test3;
-- 查询 test1 表数据
select * from hadoop_prod.default.test1;

此时test1表的数据如下：
在这里插入图片描述
我们可以看到test1表原来没有jiangsu分区，采用静态分区指定jiangsu分区的时候，并不影响非jiangsu的数据，只是从test3中读取所有数据，并存放到loc=jiangsu这个分区目录下。

注意：使用insert overwrite 读取test3表数据 静态分区方式覆盖到表 test1，表中其他分区数据不受影响，只会覆盖指定的静态分区数据。

至此，我相信我们已经完全掌握了merge into的用法。

DELETE FROM

Spark3.x版本之后支持"Delete from"可以根据指定的where条件来删除表中数据。如果where条件匹配Iceberg表一个分区的数据，Iceberg仅会修改元数据，如果where条件匹配的表的单个行，则Iceberg会只重写受影响行所在的data files。

-- 创建表 delete_tbl ,并加载数据
create table hadoop_prod.default.delete_tbl (id int,name string,age int) using iceberg;
insert into hadoop_prod.default.delete_tbl values (1,"zs",18),(2,"ls",19),(3,"ww",20),(4,"ml",21),(5,"tq",22),(6,"gb",23);
-- 根据条件范围删除表 delete_tbl 中的数据
delete from hadoop_prod.default.delete_tbl where id >3 and id <6;
-- 查询数据
select * from hadoop_prod.default.delete_tbl;

删除了id大于3和小于6之间的所有数据：
在这里插入图片描述

-- 根据条件删除表 delete_tbl 中的一条数据
delete from hadoop_prod.default.delete_tbl where id = 2;
-- 查询数据
select * from hadoop_prod.default.delete_tbl；

删除了id=2的数据：
在这里插入图片描述

删除操作和其它数据库完全一样，操作很简单，但是得理解底层删除数据的原理。

UPDATE

Spark3.x+版本支持了update更新数据操作，可以根据匹配的条件进行数据更新操作。

-- 创建表 update_tbl ,并加载数据
create table hadoop_prod.default.update_tbl (id int,name string,age int) using iceberg;
-- 插入数据
insert into hadoop_prod.default.update_tbl values (1,"zs",18),(2,"ls",19),(3,"ww",20),(4,"ml",21),(5,"tq",22),(6,"gb",23);

insert into hadoop_prod.default.update_tbl values (1,“zs”,18),(2,“ls”,19),(3,“ww”,20),(4,“ml”,21),(5,“tq”,22),(6,“gb”,23)，操作如下：

-- 更新 delete_tbl 表
update hadoop_prod.default.update_tbl set name = 'zhangsan' ,age = 30 where id <=3;
-- 查询数据
select * from hadoop_prod.default.update_tbl;

把id小于等于3的，name全部改成zhangshan，age全部改成30：
在这里插入图片描述
update操作和其它数据库一模一样，非常简单。

注意：UPDATE 更加专注于单一记录的修改，而 MERGE INTO 则是一个更全面的操作，可以同时处理多个数据状态的变化。因此一些复杂的操作直接用MERGE INTO，比如：

  • 同步外部数据源：如果你有一个外部数据库系统，你可能希望定期将更改（包括插入、更新和删除）同步到你的数据湖中的表。MERGE INTO 可以用来比较两个表，并根据匹配条件执行更新，对于没有匹配记录的新数据则执行插入。
  • 数据集成：当需要合并多个来源的数据到一个目标表中时，MERGE INTO 可以有效地处理这种情况。它可以检查数据是否已经存在，并决定是更新还是添加新的记录。
  • 高效的数据处理：在处理大量数据时，MERGE INTO 可以减少数据处理的时间，因为它只需要一次操作就可以完成更新和插入。

参考文献

Spark Write
https://bbs.huaweicloud.com/blogs/364273


http://www.ppmy.cn/devtools/109933.html

相关文章

数据库进阶：3.SQL优化

数据库进阶：3.SQL优化

1.SQL优化 1.1插入数据 insert优化 批量插入 insert into tb_test values(1,TOM),(2,caT),(3,Jerry);手动提交事务 start transaction&#xff1b; insert into tb_test values(1,Tom),(2,Cat),(3,Jerry); insert into tb_test values(1,Tom),(2,Cat),(3,Jerry); insert i…
阅读更多...
CCF刷题计划——坐标变换（其二）（前缀和）

CCF刷题计划——坐标变换（其二）（前缀和）

坐标变换&#xff08;其二&#xff09; 首先我按照一般的逻辑写出来&#xff0c;居然超时了&#xff1f;&#xff1f;&#xff1f; 之后想了想&#xff0c;还是觉得大有可为的&#xff0c;对拉伸前缀积&#xff0c;对旋转前缀和成功解决问题。 80分&#xff1a;超时 #inclu…
阅读更多...
微信小程序显示后台文章副文本，图片和视频正常显示

微信小程序显示后台文章副文本，图片和视频正常显示

解决方案: 使用 wxParse 或 rich-text 组件: 这两种方式可以解析 HTML 字符串并渲染富文本内容&#xff0c;包括图片和视频。 数据处理: 将后台返回的富文本数据进行处理&#xff0c;提取出图片和视频的链接&#xff0c;并将其转换成小程序支持的格式。 方案一&#xff1a;使…
阅读更多...
如何利用命令模式实现一个手游后端架构?

如何利用命令模式实现一个手游后端架构?

命令模式的原理解读 命令模式的英文翻译是 Command Design Pattern。在 GoF 的《设计模式》一书中&#xff0c;它是这么定义的&#xff1a; The command pattern encapsulates a request as an object, thereby letting us parameterize other objects with different reques…
阅读更多...
NVIDIA Triton Inference Server 部署 yolov5

NVIDIA Triton Inference Server 部署 yolov5

文章目录 一、拉取 tensorrt 、yolov5、tritonserver 镜像二、下载 yolov5-6.2、tensorrtx/yolov5-6.2源码三、pt转wts四、wts转engine五、创建triton推理服务器六、创建客户端进行测试 一、拉取 tensorrt 、yolov5、tritonserver 镜像 docker pull hakuyyf/tensorrtx:trt8.2_…
阅读更多...
Linux /tmp/下的文件自动清理

Linux /tmp/下的文件自动清理

在使用systemd程序的Linux系统中&#xff0c;/tmp 目录下的文件会自动清理。默认情况下&#xff0c;系统使用 systemd-tmpfiles-clean 服务来管理 /tmp 目录中的文件&#xff0c;并根据文件的访问时间来决定何时删除它们。具体清理策略取决于配置文件 /usr/lib/tmpfiles.d/tmp.…
阅读更多...
【Go】Go语言介绍与开发环境搭建

【Go】Go语言介绍与开发环境搭建

✨✨ 欢迎大家来到景天科技苑✨✨ &#x1f388;&#x1f388; 养成好习惯&#xff0c;先赞后看哦~&#x1f388;&#x1f388; &#x1f3c6; 作者简介&#xff1a;景天科技苑 &#x1f3c6;《头衔》&#xff1a;大厂架构师&#xff0c;华为云开发者社区专家博主&#xff0c;…
阅读更多...
批量创建文件夹和文件——excel VBA实现

批量创建文件夹和文件——excel VBA实现

当需要创建大量文件夹及文件时&#xff0c;可借助excel vba 实现&#xff0c;如下图&#xff1a; 批量创建文件名为1-10的文件夹&#xff0c;每个文件夹内有个与文件名相同的txt文件&#xff0c;txt文件内的数字也跟文件名相同。 附代码&#xff1a; Sub CreateFoldersAndFile…
阅读更多...
最新文章

Copyright @ 2022~2023