实践数据湖iceberg 第二十三课 flink-sql从checkpoint重启_flink sql checkpoint

系列文章目录

实践数据湖iceberg 第一课 入门
实践数据湖iceberg 第二课 iceberg基于hadoop的底层数据格式
实践数据湖iceberg 第三课 在sqlclient中，以sql方式从kafka读数据到iceberg
实践数据湖iceberg 第四课 在sqlclient中，以sql方式从kafka读数据到iceberg（升级版本到flink1.12.7）
实践数据湖iceberg 第五课 hive catalog特点
实践数据湖iceberg 第六课 从kafka写入到iceberg失败问题 解决
实践数据湖iceberg 第七课 实时写入到iceberg
实践数据湖iceberg 第八课 hive与iceberg集成
实践数据湖iceberg 第九课 合并小文件
实践数据湖iceberg 第十课 快照删除
实践数据湖iceberg 第十一课 测试分区表完整流程(造数、建表、合并、删快照)
实践数据湖iceberg 第十二课 catalog是什么
实践数据湖iceberg 第十三课 metadata比数据文件大很多倍的问题
实践数据湖iceberg 第十四课 元数据合并(解决元数据随时间增加而元数据膨胀的问题)
实践数据湖iceberg 第十五课 spark安装与集成iceberg(jersey包冲突)
实践数据湖iceberg 第十六课 通过spark3打开iceberg的认知之门
实践数据湖iceberg 第十七课 hadoop2.7,spark3 on yarn运行iceberg配置
实践数据湖iceberg 第十八课 多种客户端与iceberg交互启动命令(常用命令)
实践数据湖iceberg 第十九课 flink count iceberg，无结果问题
实践数据湖iceberg 第二十课 flink + iceberg CDC场景(版本问题，测试失败)
实践数据湖iceberg 第二十一课 flink1.13.5 + iceberg0.131 CDC(测试成功INSERT,变更操作失败)
实践数据湖iceberg 第二十二课 flink1.13.5 + iceberg0.131 CDC(CRUD测试成功)
实践数据湖iceberg 第二十三课 flink-sql从checkpoint重启

---

---

概要

问题描述：前面测试过，直接重启insert任务，下次消费，会全量数据同步一份，并进行增量的监听。多次重启的话，会有很多重复数据。 本课目的：解决这个问题，下次重启从checkpoint中恢复（测试sql的方式， jar包的方式以前测试过，不想测了）。 本课最后效果：成功测试flink-sql以非jar包的方式从checkpoint重启，完成故障恢复测试

---

1. savepoint保存

从web ui 中找到jobId, 保存
savepoint保存命令： flink savepoint {jobId} {path}

[root@hadoop101 conf]# flink savepoint 365df9287888864066b89d9f5247f654 hdfs:///tmp/savepoint
Triggering savepoint for job 365df9287888864066b89d9f5247f654.
Waiting for response...
Savepoint completed. Path: hdfs://ns/tmp/savepoint/savepoint-365df9-e6719613365a
You can resume your program from this savepoint with the run command.
[root@hadoop101 conf]# hadoop fs -ls /tmp/savepoint
Found 1 items
drwxr-xr-x   - root supergroup          0 2022-02-23 10:25 /tmp/savepoint/savepoint-365df9-e6719613365a
[root@hadoop101 conf]# hadoop fs -ls /tmp/savepoint/*
Found 1 items
-rw-r--r--   2 root supergroup       6368 2022-02-23 10:25 /tmp/savepoint/savepoint-365df9-e6719613365a/_metadata
[root@hadoop101 conf]# hadoop fs -ls /tmp/savepoint/*/*
-rw-r--r--   2 root supergroup       6368 2022-02-23 10:25 /tmp/savepoint/savepoint-365df9-e6719613365a/_metadata
[root@hadoop101 conf]# hadoop fs -du -h /tmp/savepoint/*/*
6.2 K  /tmp/savepoint/savepoint-365df9-e6719613365a/_metadata
[root@hadoop101 conf]# 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2. 制造变更数据

第一次savepoint的数据：

                                                                                                                              SQL Query Result (Table)                                                                                                                              
 Refresh: 1 s                                                                                                                     Page: Last of 1                                                                                                             Updated: 10:33:35.233 

                              i                        ts_code                         symbol                           name                           area                       industry                      list_date             actural_controller
                              1                      000002.SZ                         000002                            万科A                             深圳                           全国地产                       19910129                        星星之火!!!
                              2                      000004.SZ                         000004                           国华网安                             深圳                           软件服务                       19910114                            三体人
                              0                      000001.SZ                         000001                           平安银行                             深圳                             银行                       19910403                         (NULL)
1
2
3
4
5
6
7

制造变更：

INSERT INTO stock_basic VALUES (‘3’, ‘000005.SZ’, ‘000005’, ‘ST星源’, ‘深圳’, ‘环境保护’, ‘19901210’, ‘郑列列,丁芃’);
INSERT INTO stock_basic VALUES (‘4’, ‘000006.SZ’, ‘000006’, ‘深振业A’, ‘深圳’, ‘区域地产’, ‘19920427’, ‘深圳市人民政府国有资产监督管理委员会’);

spark-sql (default)> 
                   > select * from stock_basic_iceberg_sink;
22/02/23 10:37:59 WARN conf.HiveConf: HiveConf of name hive.metastore.event.db.notification.api.auth does not exist
i       ts_code symbol  name    area    industry        list_date       actural_controller
2       000004.SZ       000004  国华网安        深圳    软件服务        19910114        三体人
0       000001.SZ       000001  平安银行        深圳    银行    19910403        NULL
3       000005.SZ       000005  ST星源  深圳    环境保护        19901210        郑列列,丁芃
1       000002.SZ       000002  万科A   深圳    全国地产        19910129        星星之火!!!
4       000006.SZ       000006  深振业A 深圳    区域地产        19920427        深圳市人民政府国有资产监督管理委员会
Time taken: 0.478 seconds, Fetched 5 row(s)
1
2
3
4
5
6
7
8
9
10

记录savepoint

[root@hadoop101 conf]#  flink savepoint 365df9287888864066b89d9f5247f654 hdfs:///tmp/savepoint
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/module/flink-1.13.5/lib/log4j-slf4j-impl-2.16.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/module/hadoop-2.7.2/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Triggering savepoint for job 365df9287888864066b89d9f5247f654.
Waiting for response...
Savepoint completed. Path: hdfs://ns/tmp/savepoint/savepoint-365df9-6108cf790032
You can resume your program from this savepoint with the run command.
[root@hadoop101 conf]# hadoop fs -ls /tmp/savepoint
Found 2 items
drwxr-xr-x   - root supergroup          0 2022-02-23 10:42 /tmp/savepoint/savepoint-365df9-6108cf790032
drwxr-xr-x   - root supergroup          0 2022-02-23 10:25 /tmp/savepoint/savepoint-365df9-e6719613365a
[root@hadoop101 conf]# 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

第二次savepoint对应的数据

 Table program finished.                                                                                                          Page: Last of 1                                                                                                             Updated: 10:59:54.536 

                              i                        ts_code                         symbol                           name                           area                       industry                      list_date             actural_controller
                              2                      000004.SZ                         000004                           国华网安                             深圳                           软件服务                       19910114                            三体人
                              0                      000001.SZ                         000001                           平安银行                             深圳                             银行                       19910403                         (NULL)
                              1                      000002.SZ                         000002                            万科A                             深圳                           全国地产                       19910129                        星星之火!!!
                              3                      000005.SZ                         000005                           ST星源                             深圳                           环境保护                       19901210                         郑列列,丁芃
                              4                      000006.SZ                         000006                           深振业A                             深圳                           区域地产                       19920427             深圳市人民政府国有资产监督管理委员会
1
2
3
4
5
6
7
8

制造更多的变更

INSERT INTO `stock_basic` VALUES ('5', '000007.SZ', '000007', '*ST全新', '深圳', '酒店餐饮', '19920413', null);
INSERT INTO `stock_basic` VALUES ('6', '000008.SZ', '000008', '神州高铁', '北京', '运输设备', '19920507', '国家开发投资集团有限公司');

1
2
3

变更的结果

spark-sql (default)> select * from stock_basic_iceberg_sink;
22/02/23 11:04:09 WARN conf.HiveConf: HiveConf of name hive.metastore.event.db.notification.api.auth does not exist
i       ts_code symbol  name    area    industry        list_date       actural_controller
0       000001.SZ       000001  平安银行        深圳    银行    19910403        NULL
1       000002.SZ       000002  万科A   深圳    全国地产        19910129        星星之火!!!
4       000006.SZ       000006  深振业A 深圳    区域地产        19920427        深圳市人民政府国有资产监督管理委员会
5       000007.SZ       000007  *ST全新 深圳    酒店餐饮        19920413        NULL
6       000008.SZ       000008  神州高铁        北京    运输设备        19920507        国家开发投资集团有限公司
2       000004.SZ       000004  国华网安        深圳    软件服务        19910114        三体人
3       000005.SZ       000005  ST星源  深圳    环境保护        19901210        郑列列,丁芃
Time taken: 0.331 seconds, Fetched 7 row(s)
spark-sql (default)> 
1
2
3
4
5
6
7
8
9
10
11
12

3. savepoint恢复命令

准备了3个检查点

RESET execution.savepoint.path; 
SET execution.savepoint.path = 'hdfs:///tmp/savepoint/savepoint-365df9-e6719613365a;
1
2

之后执行insert 语句

RESET execution.savepoint.path; 
SET execution.savepoint.path = 'hdfs:///tmp/savepoint/savepoint-365df9-6108cf790032;

1
2
3

kill insert的job,查看checkpoint目录

[root@hadoop101 flink1.13-iceberg0131]# hadoop fs -ls /flink/checkpoints/*654/chk-17072
Found 1 items
-rw-r--r--   2 root supergroup       6407 2022-02-23 11:07 /flink/checkpoints/365df9287888864066b89d9f5247f654/chk-17072/_metadata
1
2
3

4. 从checkpoint恢复测试

下面测试 从最后的 checkpoint 点恢复

Flink SQL> 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'
> );
> 
[INFO] Execute statement succeed.

Flink SQL> 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'
> );
[INFO] Execute statement succeed.

Flink SQL> SET execution.checkpointing.interval = 3s;                                                     [INFO] Session property has been set.

Flink SQL> RESET execution.savepoint.path; 
[INFO] Session property has been reset.

Flink SQL> SET execution.savepoint.path = 'hdfs:///flink/checkpoints/365df9287888864066b89d9f5247f654/chk-17072/_metadata';
[INFO] Session property has been set.

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: 1266448a102b17131018e2c47fb8a27b


Flink SQL> select * from hive_catalog6.xxzh_stock_mysql_db.stock_basic_iceberg_sink;
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint hdfs://ns/flink/checkpoints/365df9287888864066b89d9f5247f654/chk-17072. Cannot map checkpoint/savepoint state for operator c27dcf7b54ef6bfd6cff02ca8870b681 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

insert job启动后，还是7行，数据没有翻倍。说明checkpoint恢复测试有效

spark-sql (default)> select * from stock_basic_iceberg_sink order by i;
i       ts_code symbol  name    area    industry        list_date       actural_controller
0       000001.SZ       000001  平安银行        深圳    银行    19910403        NULL
1       000002.SZ       000002  万科A   深圳    全国地产        19910129        星星之火!!!
2       000004.SZ       000004  国华网安        深圳    软件服务        19910114        三体人
3       000005.SZ       000005  ST星源  深圳    环境保护        19901210        郑列列,丁芃
4       000006.SZ       000006  深振业A 深圳    区域地产        19920427        深圳市人民政府国有资产监督管理委员会
5       000007.SZ       000007  *ST全新 深圳    酒店餐饮        19920413        NULL
6       000008.SZ       000008  神州高铁        北京    运输设备        19920507        国家开发投资集团有限公司
Time taken: 0.233 seconds, Fetched 7 row(s)
1
2
3
4
5
6
7
8
9
10

增加一行，
INSERT INTO stock_basic VALUES (‘7’, ‘000009.SZ’, ‘000009’, ‘中国宝安’, ‘深圳’, ‘电气设备’, ‘19910625’, null);
发现立即被捕捉到。

spark-sql (default)> select * from stock_basic_iceberg_sink order by i;
i       ts_code symbol  name    area    industry        list_date       actural_controller
0       000001.SZ       000001  平安银行        深圳    银行    19910403        NULL
1       000002.SZ       000002  万科A   深圳    全国地产        19910129        星星之火!!!
2       000004.SZ       000004  国华网安        深圳    软件服务        19910114        三体人
3       000005.SZ       000005  ST星源  深圳    环境保护        19901210        郑列列,丁芃
4       000006.SZ       000006  深振业A 深圳    区域地产        19920427        深圳市人民政府国有资产监督管理委员会
5       000007.SZ       000007  *ST全新 深圳    酒店餐饮        19920413        NULL
6       000008.SZ       000008  神州高铁        北京    运输设备        19920507        国家开发投资集团有限公司
7       000009.SZ       000009  中国宝安        深圳    电气设备        19910625        NULL
Time taken: 0.267 seconds, Fetched 8 row(s)
1
2
3
4
5
6
7
8
9
10
11

也就是cdc能找到恢复点，单sql-client中，直接查表，失败

5. web ui中看checkpoint恢复信息

发现本insert job的确是从 checkpoint点恢复的

---

总结

本文测试了使用 sql-client从checkpoint恢复任务。避免每次重启，重写数据