Flink Dataset Api（八）容错

迷南。 2023-05-31 10:23 13阅读 0赞

## checkpoint介绍 ##

checkpoint机制是Flink可靠性的基石，可以保证Flink集群在某个算子因为某些原因(如 异常退出)出现故障时，能够将整个应用流图的状态恢复到故障之前的某一状态，保 证应用流图状态的一致性。Flink的checkpoint机制原理来自“Chandy-Lamport algorithm”算法。

每个需要checkpoint的应用在启动时，Flink的JobManager为其创建一个 CheckpointCoordinator，CheckpointCoordinator全权负责本应用的快照制作

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

1. CheckpointCoordinator周期性的向该流应用的所有source算子发送barrier。 
    2.当某个source算子收到一个barrier时，便暂停数据处理过程，然后将自己的当前状 态制作成快照，并保存到
    指定的持久化存储中，最后向CheckpointCoordinator报告 自己快照制作情况，同时向自身所有下游算子广播该
    barrier，恢复数据处理 
    3.下游算子收到barrier之后，会暂停自己的数据处理过程，然后将自身的相关状态制作成快照，并保存到指定的
    持久化存储中，最后向CheckpointCoordinator报告自身 快照情况，同时向自身所有下游算子广播该barrier，
    恢复数据处理。 
    4. 每个算子按照步骤3不断制作快照并向下游广播，直到最后barrier传递到sink算子，快照制作完成。 
    5. 当CheckpointCoordinator收到所有算子的报告之后，认为该周期的快照制作成功; 否则，如果在规定的时间
    内没有收到所有算子的报告，则认为本周期快照制作失败

如果一个算子有两个输入源，则暂时阻塞先收到barrier的输入源，等到第二个输入源相 同编号的barrier到来时，再制作自身快照并向下游广播该barrier。具体如下图所示

两个输入源 checkpoint 过程 ：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

1. 假设算子C有A和B两个输入源 
    2. 在第i个快照周期中，由于某些原因(如处理时延、网络时延等)输入源A发出的 barrier先到来，这时算子C暂时将输入源A的输入通道阻塞，仅收输入源B的数据。 
    3. 当输入源B发出的barrier到来时，算子C制作自身快照并向CheckpointCoordinator报 告自身的快照制作情况，然后将两个barrier合并为一个，向下游所有的算子广播。

当由于某些原因出现故障时，CheckpointCoordinator通知流图上所有算子统一恢复到某 个周期的checkpoint状态，然后恢复数据流处理。分布式checkpoint机制保证了数据仅被 处理一次(Exactly Once)。

## 持久化存储 ##

目前，Checkpoint持久化存储可以使用如下三种:

MemStateBackend

该持久化存储主要将快照数据保存到JobManager的内存中，仅适合作为测试以及
    
    快照的数据量非常小时使用，并不推荐用作大规模商业部署。

FsStateBackend

该持久化存储主要将快照数据保存到文件系统中，目前支持的文件系统主要是 HDFS和本地文件。
    如果使用HDFS，则初始化FsStateBackend时，需要传入以 “hdfs://”开头的路径(即: new FsStateBackend("hdfs:///hacluster/checkpoint"))， 
    如果使用本地文件，则需要传入以“file://”开头的路径(即:new FsStateBackend("file:///Data"))。
    在分布式情况下，不推荐使用本地文件。如果某 个算子在节点A上失败，在节点B上恢复，
    使用本地文件时，在B上无法读取节点 A上的数据，导致状态恢复失败。

RocksDBStateBackend

RocksDBStatBackend介于本地文件和HDFS之间，平时使用RocksDB的功能，将数 据持久化到本地文件中，当制作
    快照时，将本地数据制作成快照，并持久化到 FsStateBackend中(FsStateBackend不必用户特别指明，只需在初
    始化时传入HDFS 或本地路径即可，如new RocksDBStateBackend("hdfs:///hacluster/checkpoint")或new 
    RocksDBStateBackend("file:///Data"))。
    
    如果用户使用自定义窗口(window)，不推荐用户使用RocksDBStateBackend。在自 定义窗口中，状态以
    ListState的形式保存在StatBackend中，如果一个key值中有多 个value值，则RocksDB读取该种ListState非常
    缓慢，影响性能。用户可以根据应用 的具体情况选择FsStateBackend+HDFS或RocksStateBackend+HDFS。

语法：

val env = StreamExecutionEnvironment.getExecutionEnvironment
        // start a checkpoint every 1000 ms
        env.enableCheckpointing(1000)
        // advanced options:
        // 设置checkpoint的执行模式，最多执行一次或者至少执行一次
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        // 设置checkpoint的超时时间
        env.getCheckpointConfig.setCheckpointTimeout(60000)
        // 如果在只做快照过程中出现错误，是否让整体任务失败：true是  false不是
        env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
        //设置同一时间有多少 个checkpoint可以同时执行 
        env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

需求：

假定用户需要每隔1秒钟需要统计4秒中窗口中数据的量，然后对统计的结果值进行checkpoint处理

数据规划：

\\1. 使用自定义算子每秒钟产生大约10000条数据。

\\2. 产生的数据为一个四元组(Long，String，String，Integer)—------(id,name,info,count)。

\\3. 数据经统计后，统计结果打印到终端输出。

\\4. 打印输出的结果为Long类型的数据。

开发思路：

\\1. source算子每隔1秒钟发送10000条数据，并注入到Window算子中。

\\2. window算子每隔1秒钟统计一次最近4秒钟内数据数量。

\\3. 每隔1秒钟将统计结果打印到终端

\\4. 每隔6秒钟触发一次checkpoint，然后将checkpoint的结果保存到HDFS中。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: /images/20230531/c7a1faae354c432192bb275504778581.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: /images/20230531/7b45fb6d9d2847e998b0fc494eb3050b.png