概要
版本：flink1.13.5, flink-sql-connector-mysql-cdc-2.1.1.jar ,iceberg0.131
本课：测试cdc的CRUD，顺利测试通过insert,delete,update，CRUD都通过。
重点：CRUD测试通过的关键在于把sink 到iceberg表的version设置为2

一、环境准备
1.1 准备安装包、jar包
flink-1.13.5-bin-scala_2.12.tgz
解压
采用软连接方式进行安装，每次版本升级，把软连接更换，把老版本配置/conf下的文件拷贝到新的路径就行。
这样的好处，环境变量不用每次都修改。
看版本历史，已经试过了4个版本的flink，摸索的艰辛泪。。。

[root@hadoop101 module]# ll
total 94528
drwxr-xr-x 11 hadoop hadoop 4096 Jan 11 17:52 apache-hive-2.3.6-bin
drwxr-xr-x 2 root root 4096 Feb 14 18:25 bin
lrwxrwxrwx 1 root root 25 Feb 17 21:30 flink -> /opt/module/flink-1.13.5/
drwxr-xr-x 10 hadoop hadoop 4096 Jan 12 15:03 flink-1.11.6
drwxr-xr-x 10 1002 1003 4096 Dec 15 08:30 flink-1.12.7
drwxr-xr-x 10 1006 1007 4096 Dec 15 08:35 flink-1.13.5
drwxr-xr-x 10 501 games 4096 Jan 11 07:45 flink-1.14.3

准备：flink集成kafka, hive, iceberg的包

[root@hadoop101 module]# ls /opt/software/flink1.13-iceberg0131/
flink-sql-connector-hive-2.3.6_2.12-1.13.5.jar flink-sql-connector-kafka_2.12-1.13.5.jar iceberg-flink-runtime-1.13-0.13.1.jar iceberg-mr-0.13.1.jar
flink-sql-connector-mysql-cdc-2.1.1.jar’
1
2
3
包从 https://repo.maven.apache.org/maven2/org/apache 目录下直接找就是了。
mysql-cdc 自己编译，maven仓库down的是scala2.11版本，我用的是scala.2.12版，不兼容（踩过坑了）。

[root@hadoop103 target]# pwd
/opt/software/flink-cdc-connectors-release-2.1.1/flink-sql-connector-mysql-cdc/target
[root@hadoop103 target]# ls
checkstyle-checker.xml checkstyle-suppressions.xml dependency-reduced-pom.xml flink-sql-connector-mysql-cdc-2.1.1-tests.jar maven-archiver maven-status test-classes
checkstyle-result.xml classes flink-sql-connector-mysql-cdc-2.1.1.jar generated-sources maven-shared-archive-resources original-flink-sql-connector-mysql-cdc-2.1.1.jar
[root@hadoop103 target]#

1.2 flink-sql启动
[root@hadoop101 ~]# sql-client.sh embedded -j /opt/software/flink1.13-iceberg0131/iceberg-flink-runtime-1.13-0.13.1.jar -j /opt/software/flink1.13-iceberg0131/flink-sql-connector-hive-2.3.6_2.12-1.13.5.jar -j /opt/software/flink1.13-iceberg0131/flink-sql-connector-kafka_2.12-1.13.5.jar -j /opt/software/flink1.13-iceberg0131/flink-sql-connector-mysql-cdc-2.1.1.jar shell

2.1 准备mysql的表source表
CREATE TABLE stock_basic_source(
`i` INT NOT NULL,
`ts_code` CHAR(10) NOT NULL,
`symbol` CHAR(10) NOT NULL,
`name` char(10) NOT NULL,
`area` CHAR(20) NOT NULL,
`industry` CHAR(20) NOT NULL,
`list_date` CHAR(10) NOT NULL,
`actural_controller` CHAR(100),
PRIMARY KEY(i) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'hadoop103',
'port' = '3306',
'username' = 'hive',
'password' = '123456',
'database-name' = 'xxzh_stock',
'table-name' = 'stock_basic'
);

先执行如下8条：
INSERT INTO `stock_basic` VALUES ('0', '000001.SZ', '000001', '平安银行', '深圳', '银行', '19910403', null);
INSERT INTO `stock_basic` VALUES ('1', '000002.SZ', '000002', '万科A', '深圳', '全国地产', '19910129', null);
INSERT INTO `stock_basic` VALUES ('2', '000004.SZ', '000004', '国华网安', '深圳', '软件服务', '19910114', '李映彤');
INSERT INTO `stock_basic` VALUES ('3', '000005.SZ', '000005', 'ST星源', '深圳', '环境保护', '19901210', '郑列列,丁芃');
INSERT INTO `stock_basic` VALUES ('4', '000006.SZ', '000006', '深振业A', '深圳', '区域地产', '19920427', '深圳市人民政府国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('5', '000007.SZ', '000007', '*ST全新', '深圳', '酒店餐饮', '19920413', null);
INSERT INTO `stock_basic` VALUES ('6', '000008.SZ', '000008', '神州高铁', '北京', '运输设备', '19920507', '国家开发投资集团有限公司');
INSERT INTO `stock_basic` VALUES ('7', '000009.SZ', '000009', '中国宝安', '深圳', '电气设备', '19910625', null);


INSERT INTO `stock_basic` VALUES ('8', '000010.SZ', '000010', '美丽生态', '深圳', '建筑工程', '19951027', '沈玉兴');
INSERT INTO `stock_basic` VALUES ('9', '000011.SZ', '000011', '深物业A', '深圳', '区域地产', '19920330', '深圳市人民政府国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('10', '000012.SZ', '000012', '南玻A', '深圳', '玻璃', '19920228', null);
INSERT INTO `stock_basic` VALUES ('11', '000014.SZ', '000014', '沙河股份', '深圳', '全国地产', '19920602', '深圳市人民政府国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('12', '000016.SZ', '000016', '深康佳A', '深圳', '家用电器', '19920327', '国务院国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('13', '000017.SZ', '000017', '深中华A', '深圳', '文教休闲', '19920331', null);
INSERT INTO `stock_basic` VALUES ('14', '000019.SZ', '000019', '深粮控股', '深圳', '其他商业', '19921012', '深圳市人民政府国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('15', '000020.SZ', '000020', '深华发A', '深圳', '元器件', '19920428', '李中秋');
INSERT INTO `stock_basic` VALUES ('16', '000021.SZ', '000021', '深科技', '深圳', 'IT设备', '19940202', '中国电子信息产业集团有限公司');
INSERT INTO `stock_basic` VALUES ('17', '000023.SZ', '000023', '深天地A', '深圳', '水泥', '19930429', '林宏润');
INSERT INTO `stock_basic` VALUES ('18', '000025.SZ', '000025', '特力A', '深圳', '汽车服务', '19930621', '深圳市人民政府国有资产监督管理委员会');
INSERT INTO `stock_basic` VALUES ('19', '000026.SZ', '000026', '飞亚达', '深圳', '其他商业', '19930603', '中国航空技术国际控股有限公司');
INSERT INTO `stock_basic` VALUES ('20', '000027.SZ', '000027', '深圳能源', '深圳', '火力发电', '19930903', '深圳市人民政府国有资产监督管理委员会');

2.2 准备iceberg的sink表
重要配置

设置metadata保留次数
‘format-version’=‘2’，配置了才支持delete和update

CREATE CATALOG hive_catalog6 WITH (
'type'='iceberg',
'catalog-type'='hive',
'uri'='thrift://hadoop101:9083',
'clients'='5',
'property-version'='1',
'warehouse'='hdfs:///user/hive/warehouse/hive_catalog6'
);
use catalog hive_catalog6;

CREATE DATABASE xxzh_stock_mysql_db;
USE xxzh_stock_mysql_db;

CREATE TABLE stock_basic_iceberg_sink(
`i` INT NOT NULL,
`ts_code` CHAR(10) NOT NULL,
`symbol` CHAR(10) NOT NULL,
`name` char(10) NOT NULL,
`area` CHAR(20) NOT NULL,
`industry` CHAR(20) NOT NULL,
`list_date` CHAR(10) NOT NULL,
`actural_controller` CHAR(100) ,
PRIMARY KEY(i) NOT ENFORCED
) with(
'write.metadata.delete-after-commit.enabled'='true',
'write.metadata.previous-versions-max'='5',
'format-version'='2'
)


给表增加属性的方法：

create table tablename(
field1 field_type
) with (
'key' = 'value'
)

3.通过flink从mysql写入iceberg
3.1 数据写入到iceberg
use catalog default_catalog;

设置checkpoint频率为3s，（不设置checkpont，是没有写操作的），写在公共配置也可以

Flink SQL> SET execution.checkpointing.interval = 3s;
1
Flink SQL> insert into hive_catalog6.xxzh_stock_mysql_db.stock_basic_iceberg_sink select * from stock_basic_source;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 3ed67670f3f008e19409e99d781d92d3

4. 观察增加、 删除、更新的影响
4.1 观察方法：flink-sql中select
Flink SQL> select * from stock_basic_source
1
4.2 测试delete（通过）
mysql上执行： delete from stock_basic;
发现表清空了


4.3 测试insert（通过）
INSERT INTO `stock_basic` VALUES ('0', '000001.SZ', '000001', '平安银行', '深圳', '银行', '19910403', null);
1


INSERT INTO `stock_basic` VALUES ('1', '000002.SZ', '000002', '万科A', '深圳', '全国地产', '19910129', null);
1


4.4 测试update（通过）
update stock_basic set actural_controller='星星之火' where i='1'
1


update stock_basic set actural_controller='星星之火!!!' where i='1'
1


5 hdfs文件观察
5.1 hdfs文件内容
[root@hadoop101 conf]# hadoop fs -ls hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata
Found 43 items
-rw-r--r-- 2 root supergroup 16197 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00015-30461c7a-82f8-4f33-a3a7-de69d16284b9.metadata.json
-rw-r--r-- 2 root supergroup 17224 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00016-d735a8b9-8ba8-43cd-bfe0-28582ec20857.metadata.json
-rw-r--r-- 2 root supergroup 18080 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00017-810ca00f-5006-4794-ab71-8c239ea5f7f9.metadata.json
-rw-r--r-- 2 root supergroup 18936 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00018-5cb08dca-cb02-453c-a901-1c573aa7ffc8.metadata.json
-rw-r--r-- 2 root supergroup 19963 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00019-a0638ba0-3199-4ed8-af20-9041b16f49e7.metadata.json
-rw-r--r-- 2 root supergroup 20819 2022-02-22 21:02 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/00020-d41f56dc-726d-40f1-842a-01dc0943fb08.metadata.json
-rw-r--r-- 2 root supergroup 7014 2022-02-22 20:58 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/2bf52ac8-356b-4472-95ce-36652483f4da-m0.avro
-rw-r--r-- 2 root supergroup 7021 2022-02-22 17:25 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/3ed67670f3f008e19409e99d781d92d3-00000-0-23-00003.avro
-rw-r--r-- 2 root supergroup 7021 2022-02-22 17:25 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/3ed67670f3f008e19409e99d781d92d3-00000-0-23-00004.avro
-rw-r--r-- 2 root supergroup 7035 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/53b9c6dc-b0f6-494d-bed4-810dcaaabdc8-m0.avro
-rw-r--r-- 2 root supergroup 7026 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/53b9c6dc-b0f6-494d-bed4-810dcaaabdc8-m1.avro
-rw-r--r-- 2 root supergroup 7022 2022-02-22 20:59 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/65d768c5-9242-4b59-a8ae-19f4f3d768cc-m0.avro
-rw-r--r-- 2 root supergroup 7013 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/68a8c6ac-af79-47c6-8898-5f72af09fc48-m0.avro
-rw-r--r-- 2 root supergroup 7021 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/8b812160-dc0d-4c61-939f-c7ed3f263ea0-m0.avro
-rw-r--r-- 2 root supergroup 7099 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/8c28b913-80d1-4fd7-97d8-77cac91c86c9-m0.avro
-rw-r--r-- 2 root supergroup 7039 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/c299d8ed-513b-4144-aab1-a338622a8344-m0.avro
-rw-r--r-- 2 root supergroup 7039 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/c299d8ed-513b-4144-aab1-a338622a8344-m1.avro
-rw-r--r-- 2 root supergroup 7112 2022-02-22 18:11 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/c7fe51e05881714e4e2fa8bf2243fff2-00000-0-6-00003.avro
-rw-r--r-- 2 root supergroup 7033 2022-02-22 18:11 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/c7fe51e05881714e4e2fa8bf2243fff2-00000-0-6-00004.avro
-rw-r--r-- 2 root supergroup 7095 2022-02-22 20:54 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/d31b1088-b459-4120-b5fe-b110c9140a2f-m0.avro
-rw-r--r-- 2 root supergroup 7116 2022-02-22 18:02 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/edd981ba1e94116b0087da599b803bcd-00000-0-5-00002.avro
-rw-r--r-- 2 root supergroup 7018 2022-02-22 18:02 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/edd981ba1e94116b0087da599b803bcd-00000-0-5-00003.avro
-rw-r--r-- 2 root supergroup 7063 2022-02-22 20:57 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/fb272824-1ef7-4b0b-ac2f-06596a231b4d-m0.avro
-rw-r--r-- 2 root supergroup 4361 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-1642441739864897840-1-8c28b913-80d1-4fd7-97d8-77cac91c86c9.avro
-rw-r--r-- 2 root supergroup 4408 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-1942818859034480839-1-68a8c6ac-af79-47c6-8898-5f72af09fc48.avro
-rw-r--r-- 2 root supergroup 4287 2022-02-22 20:54 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-308385720639088133-1-d31b1088-b459-4120-b5fe-b110c9140a2f.avro
-rw-r--r-- 2 root supergroup 4556 2022-02-22 20:58 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-4299157873725018231-1-2bf52ac8-356b-4472-95ce-36652483f4da.avro
-rw-r--r-- 2 root supergroup 4599 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-4674301170568153452-1-749d02e4-9949-4f10-8153-bbb57fcfead2.avro
-rw-r--r-- 2 root supergroup 4706 2022-02-22 21:02 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-5046003445908931550-1-26c22e73-f906-49a6-a729-26a9a2621fbe.avro
-rw-r--r-- 2 root supergroup 4360 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-547797696930809711-1-31916bf9-3e25-40d4-97f6-d5130dea38e6.avro
-rw-r--r-- 2 root supergroup 4302 2022-02-22 20:54 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-5920052446851138250-1-64fff5de-1939-426f-81f3-2f06e23dd5dd.avro
-rw-r--r-- 2 root supergroup 4658 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-5935626013158511935-1-53b9c6dc-b0f6-494d-bed4-810dcaaabdc8.avro
-rw-r--r-- 2 root supergroup 4556 2022-02-22 20:59 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-5937571809584377788-1-b2337381-7c49-4400-ad73-d202c11725d9.avro
-rw-r--r-- 2 root supergroup 4658 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-6113597286018236924-1-194ca1a1-7c40-4566-9a82-71669901ab02.avro
-rw-r--r-- 2 root supergroup 4454 2022-02-22 20:56 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-616292368608451816-1-935de70b-3afe-4234-92c8-fe5f4fb8a14b.avro
-rw-r--r-- 2 root supergroup 4505 2022-02-22 20:57 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-6177965649138968933-1-fb272824-1ef7-4b0b-ac2f-06596a231b4d.avro
-rw-r--r-- 2 root supergroup 4658 2022-02-22 21:00 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-6435980392319646015-1-fce4dd40-c68a-4dee-8ece-d454d1ded33e.avro
-rw-r--r-- 2 root supergroup 4599 2022-02-22 20:59 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-7339825486123029008-1-65d768c5-9242-4b59-a8ae-19f4f3d768cc.avro
-rw-r--r-- 2 root supergroup 4454 2022-02-22 20:57 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-7933766380439735607-1-51420c1e-eaf6-41d5-b2d0-0d1992279039.avro
-rw-r--r-- 2 root supergroup 4455 2022-02-22 20:55 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-8562748907415457874-1-8b812160-dc0d-4c61-939f-c7ed3f263ea0.avro
-rw-r--r-- 2 root supergroup 4556 2022-02-22 20:58 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-8831941079954378156-1-eab804cf-3886-4874-a99f-8910c727048b.avro
-rw-r--r-- 2 root supergroup 4455 2022-02-22 20:56 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-8877554487080654834-1-335b29c3-f1b6-44ea-8b4a-3075863a1f45.avro
-rw-r--r-- 2 root supergroup 4706 2022-02-22 21:01 hdfs://ns/user/hive/warehouse/xxzh_stock_mysql_db.db/stock_basic_iceberg_sink/metadata/snap-9056764754290690961-1-c299d8ed-513b-4144-aab1-a338622a8344.avro
[root@hadoop101 conf]#

5.2 文件分析
metadata.json的个数是6个，data文件很多个。符合预期。
每次insert、delete、update在数据写时，都生成文件，metadata.json维持6个，是由表定义中决定。

总结
通过设置表的版本号为2，才能支持delete和update，默认是version=1，是不支持,delete 、update的。

内容来源于网络，如侵删。

扫码下载《数据治理行业实践白皮书》，下载地址：https://fs80.cn/4w2atu

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack