前言

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 ...) --匹配不上向目标表插入数据

2. 示例

• 创建两张表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