FlinkX断点续传原理

本文主要是介绍FlinkX断点续传原理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

FlinkX断点续传原理

声明： 前半部分官方文档+贴实现代码

前提条件

同步任务要支持断点续传，对数据源有一些强制性的要求：

• 数据源（这里特指关系数据库）中必须包含一个升序的字段，比如主键或者日期类型的字段，同步过程中会使用checkpoint机制记录这个字段的值，任务恢复运行时使用这个字段构造查询条件过滤已经同步过的数据，如果这个字段的值不是升序的，那么任务恢复时过滤的数据就是错误的，最终导致数据的缺失或重复；
• 数据源必须支持数据过滤，如果不支持的话，任务就无法从断点处恢复运行，会导致数据重复；
• 目标数据源必须支持事务，比如关系数据库，文件类型的数据源也可以通过临时文件的方式支持；

配置

restore用于配置同步任务类型（离线同步、实时采集）和断点续传功能。具体配置如下所示：

"restore" : {"isStream" : false,"isRestore" : false,"restoreColumnName" : "","restoreColumnIndex" : 0,"maxRowNumForCheckpoint" : 10000
}

名称	说明	是否必填	默认值	参数类型
isStream	是否为实时采集任务	否	false	Boolean
isRestore	是否开启断点续传	否	false	Boolean
restoreColumnName	断点续传字段名称	开启断点续传后必填	无	String
restoreColumnIndex	断点续传字段索引ID	开启断点续传后必填	-1	int
maxRowNumForCheckpoint	触发checkpoint数据条数	否	10000	int

任务运行的详细过程

我们用一个具体的任务详细介绍一下整个过程，任务详情如下：

数据源	mysql表，假设表名data_test，表中包含主键字段id
目标数据源	hdfs文件系统，假设写入路径为 /data_test
并发数	2
checkpoint配置	时间间隔为60s，checkpoint的StateBackend为FsStateBackend，路径为 /flinkx/checkpoint
jobId	用来构造数据文件的名称，假设为 abc123

1) 读取数据

读取数据时首先要构造数据分片，构造数据分片就是根据通道索引和checkpoint记录的位置构造查询sql，sql模板如下：

select * from data_test
where id mod ${channel_num}=${channel_index}
and id > ${offset}

如果是第一次运行，或者上一次任务失败时还没有触发checkpoint，那么offset就不存在，根据offset和通道可以确定具体的查询sql:

offset存在时

第一个通道：
select * from data_test
where id mod 2=0
and id > ${offset_0};第二个通道
select * from data_test
where id mod 2=1
and id > ${offset_1};

offset不存在时

第一个通道：
select * from data_test
where id mod 2=0;第二个通道
select * from data_test
where id mod 2=1;

数据分片构造好之后，每个通道就根据自己的数据分片去读数据了。

2）写数据

写数据前会先做几个操作：

1. 检测 /data_test 目录是否存在，如果目录不存在，则创建这个目录，如果目录存在，进行2操作；

2. 判断是不是以覆盖模式写数据，如果是，则删除 /data_test目录，然后再创建目录，如果不是，则进行3操作;

3. 检测 /data_test/.data 目录是否存在，如果存在就先删除，再创建，确保没有其它任务因异常失败遗留的脏数据文件；

   数据写入hdfs是单条写入的，不支持批量写入。数据会先写入/data_test/.data/目录下，数据文件的命名格式为： channelIndex.jobId.fileIndex，包含通道索引，jobId，文件索引三个部分，文件最掐灭。

这篇关于FlinkX断点续传原理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---