FlinkX断点续传原理

妖狐艹你老母 2023-02-13 03:21 3阅读 0赞

# FlinkX断点续传原理 #

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

## **前提条件** ##

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

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

## 配置 ##

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

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

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

## **任务运行的详细过程** ##

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

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

### **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，文件索引三个部分，文件最掐灭。

### **3**）checkpoint触发时 ###

在FlinkX中“状态”表示的是标识字段id的值，我们假设checkpoint触发时两个通道的读取和写入情况如图中所示：

![image-20200524202934601][]

checkpoint触发后，两个reader先生成Snapshot记录读取状态，通道0的状态为 id=12，通道1的状态为 id=11。Snapshot生成之后向数据流里面插入barrier，barrier随数据流向Writer。以Writer\_0为例，以Writer\_0接收Reader\_0和Reader\_1的发来的数据，假设先收到了Reader\_0的barrier，这个时候Writer\_0停止写出数据到HDFS，将接收到的数据先放到 InputBuffer里面，一直等待Reader\_1的barrier到达之后再将Buffer里的数据全部写出，然后生成Writer的Snapshot，整个checkpoint结束后，记录的任务状态为：

> Reader\_0：id=12
> 
> Reader\_1：id=11
> 
> Writer\_0：id=无法确定
> 
> Writer\_1：id=无法确定

任务状态会记录到配置的HDFS目录/flinkx/checkpoint/abc123下。因为每个Writer会接收两个Reader的数据，以及各个通道的数据读写速率可能不一样，所以导致writer接收到的数据顺序是不确定的，但是这不影响数据的准确性，因为读取数据时只需要Reader记录的状态就可以构造查询sql，我们只要确保这些数据真的写到HDF就行了。在Writer生成Snapshot之前，会做一系列操作保证接收到的数据全部写入HDFS：

*  close写入HDFS文件的数据流，这时候会在/data\_test/.data目录下生成两个两个文件:
    
     *  /data\_test/.data/0.abc123.0
     *  /data\_test/.data/1.abc123.0
 *  将生成的两个数据文件移动到/data\_test目录下；
 *  更新文件名称模板更新为：channelIndex.abc123.1;

快照生成后任务继续读写数据，如果生成快照的过程中有任何异常，任务会直接失败，这样这次快照就不会生成，任务恢复时会从上一个成功的快照恢复。

### **4**）任务正常结束 ###

任务正常结束时也会做和生成快照时同样的操作，close文件流，移动临时数据文件等。

### **5**）任务异常终止 ###

任务如果异常结束，假设任务结束时最后一个checkpoint记录的状态为：

> Reader\_0：id=12
> 
> Reader\_1：id=11

那么任务恢复的时候就会把各个通道记录的状态赋值给offset，再次读取数据时构造的sql为：

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

这样就可以从上一次失败的位置继续读取数据了。

## 代码实现 ##

### 全局checkpoint配置 ###

private static StreamExecutionEnvironment openCheckpointConf(StreamExecutionEnvironment env, Properties properties){ 
            if(properties!=null){ 
                String interval = properties.getProperty(ConfigConstant.FLINK_CHECKPOINT_INTERVAL_KEY);
                if(StringUtils.isNotBlank(interval)){ 
                    env.enableCheckpointing(Long.parseLong(interval.trim()));
                    LOG.info("Open checkpoint with interval:" + interval);
                }
                String checkpointTimeoutStr = properties.getProperty(ConfigConstant.FLINK_CHECKPOINT_TIMEOUT_KEY);
                if(checkpointTimeoutStr != null){ 
                    long checkpointTimeout = Long.parseLong(checkpointTimeoutStr.trim());
                    //checkpoints have to complete within one min,or are discard
                    env.getCheckpointConfig().setCheckpointTimeout(checkpointTimeout);
    
                    LOG.info("Set checkpoint timeout:" + checkpointTimeout);
                }
                env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
                env.getCheckpointConfig().enableExternalizedCheckpoints(
                        CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            }
            return env;
        }

*  interval 参数： `flink.checkpoint.interval` 每次checkpoint打快照间隔时间
 *  timeout 参数： `flink.checkpoint.timeout` 每次checkpoint打快照过程的超时时间
 *  checkpoint 模式： 恰好一次语义
 *  任务取消后，需要手动删除状态。

### 状态管理（CheckpointFunction） ###

`DtInputFormatSourceFunction` 所有的Reader 都会加上这个DT 包装，官方未写明实际作用

@Override
    	public void snapshotState(FunctionSnapshotContext context) throws Exception { 
            FormatState formatState = ((BaseRichInputFormat) format).getFormatState();
            if (formatState != null){ 
                LOG.info("InputFormat format state:{}", formatState.toString());
                unionOffsetStates.clear();
                unionOffsetStates.add(formatState);
            }
    	}
    
    	@Override
    	public void initializeState(FunctionInitializationContext context) throws Exception { 
    	    LOG.info("Start initialize input format state");
            // 获取OperatorStateStore
    		OperatorStateStore stateStore = context.getOperatorStateStore();
            
            // 获取全局统一的State
            unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(
    				LOCATION_STATE_NAME,
    				TypeInformation.of(new TypeHint<FormatState>() { })));
    
            LOG.info("Is restored:{}", context.isRestored());
            // 如果当前上下文是恢复状态
    		if (context.isRestored()){ 
    			formatStateMap = new HashMap<>(16);
    			for (FormatState formatState : unionOffsetStates.get()) { 
    				formatStateMap.put(formatState.getNumOfSubTask(), formatState);
    				LOG.info("Input format state into:{}", formatState.toString());
    			}
    		}
    
            LOG.info("End initialize input format state");
    	}

1.  当checkpoint被触发的时候，保存 `formatState`
2.  当恢复时，从checkpoint中恢复反序列化出`formatState`
3.  以`JDBC`为例

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70][]

### 状态恢复 ###

然后在`InputFormatSourceFunction` 重新启动的时候，从`StateBackend`恢复刚刚的`FormatState`,并设置回去。

![image-20200808144710906][]

**以 `JDBC`为例**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70 1][]

### 状态类 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70 2]

### 配置Flink任务的重启策略 ###

private static void configRestartStrategy(StreamExecutionEnvironment env, DataTransferConfig config){ 
            if (needRestart(config)) { 
                RestartConfig restartConfig = findRestartConfig(config);
                if (RestartConfig.STRATEGY_FIXED_DELAY.equalsIgnoreCase(restartConfig.getStrategy())) { 
                    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                            restartConfig.getRestartAttempts(),
                            Time.of(restartConfig.getDelayInterval(), TimeUnit.SECONDS)
                    ));
                } else if (RestartConfig.STRATEGY_FAILURE_RATE.equalsIgnoreCase(restartConfig.getStrategy())) { 
                    env.setRestartStrategy(RestartStrategies.failureRateRestart(
                            restartConfig.getFailureRate(),
                            Time.of(restartConfig.getFailureInterval(), TimeUnit.SECONDS),
                            Time.of(restartConfig.getDelayInterval(), TimeUnit.SECONDS)
                    ));
                } else { 
                    env.setRestartStrategy(RestartStrategies.noRestart());
                }
            }
        }

策略：

*  固定时间
 *  错误率
 *  无需重启

## 支持断点续传的插件 ##

理论上只要支持过滤数据的数据源，和支持事务的数据源都可以支持断点续传的功能，目前FlinkX支持的插件如下：

<table> 
 <thead> 
  <tr> 
   <th>Reader</th> 
   <th>Writer</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mysql等关系数据读取插件</td> 
   <td>HDFS</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>FTP</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>mysql等关系数据库写入插件</td> 
  </tr> 
 </tbody> 
</table>

[image-20200524202934601]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjQyMDI5MzQ2MDEucG5n?x-oss-process=image/format,png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200808150726128.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ=,size_16,color_FFFFFF,t_70
[image-20200808144710906]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA4MDgxNDQ3MTA5MDYucG5n?x-oss-process=image/format,png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200808150711481.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/2020080815070196.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ=,size_16,color_FFFFFF,t_70