当前位置:   article > 正文

Flink笔记17:检查点(checkpoint)、保存点(save points)与重启策略配置_flink savepoint会先做一次checkpoint吗

flink savepoint会先做一次checkpoint吗

1、检查点(checkpoint)

Flink 故障恢复机制的核心,就是应用状态的一致性检查点。有状态流应用的一致检查点,其实就是所有任务的状态,在某个时间点的一份拷贝(一份快照);这个时间点,应该是所有任务都恰好处理完一个相同的输入数据的时候。

下图是一个奇偶求和的例子,sum_even表示偶数求和,sum_odd表示奇数求和,数据源是1-9的数字。从图中可以看出,source偏移量为5,表示已经读取了1-5的数字,sum_even=2+4=6,sum_odd=1+3+5=9。此时做一个checkpoint,将状态保存到状态后端。
在这里插入图片描述
从检查点恢复状态
在这里插入图片描述
继续读取数据,当读取到数字7时,计算奇数和的程序还没来得及加上7就挂了,此时偶数和为12。
在这里插入图片描述
第一步重启应用,此时Source、sum_even和sum_odd的状态都被清空。
在这里插入图片描述
第二步是从 checkpoint 中读取状态,将状态重置。从检查点重新启动应用程序后,其内部状态与检查点完成时的状态完全相同。
在这里插入图片描述
第三步:开始消费并处理检查点到发生故障之间的所有数据(就是把6和7加上)。这种检查点的保存和恢复机制可以为应用程序状态提供“精确一次”(exactlyonce)的一致性,因为所有算子都会保存检查点并恢复其所有状态,这样一来所有的输入流就都会被重置到检查点完成时的位置。

2、检查点算法

  • 一种简单的想法
    暂停应用,保存状态到检查点,再重新恢复应用
  • Flink 的改进实现
    基于 Chandy-Lamport 算法的分布式快照
    将检查点的保存和数据处理分离开,不暂停整个应用

检查点分界线(Checkpoint Barrier)

Flink 的检查点算法用到了一种称为分界线(barrier)的特殊数据形式,用来把一条流上数据按照不同的检查点分开。分界线之前到来的数据导致的状态更改,都会被包含在当前分界线所属的检查点中;而基于分界线之后的数据导致的所有更改,就会被包含在之后的检查点中。

算法举例说明:
在这里插入图片描述
现在是一个有两个输入流的应用程序,用并行的两个 Source 任务来读取。两个Source 读取的都是1-9的数字。
在这里插入图片描述
JobManager 会向每个 source 任务发送一条带有新检查点 ID 的消息,通过这种方式来启动检查点。
在这里插入图片描述
数据源将它们的状态写入检查点,并发出一个检查点 barrier。状态后端在状态存入检查点之后,会返回通知给 source 任务,source 任务就会向 JobManager 确认检查点完成。
在这里插入图片描述
分界线对齐:barrier 向下游传递,sum 任务会等待所有输入分区的 barrier 到达。对于barrier已经到达的分区,继续到达的数据会被缓存;而barrier尚未到达的分区,数据会被正常处理。
在这里插入图片描述
当收到所有输入分区的 barrier 时,任务就将其状态保存到状态后端的检查点中,然后将 barrier 继续向下游转发。
在这里插入图片描述
向下游转发检查点 barrier 后,任务继续正常的数据处理。
在这里插入图片描述
Sink 任务向 JobManager 确认状态保存到 checkpoint 完毕。当所有任务都确认已成功将状态保存到检查点时,检查点就真正完成了。

3、代码中配置checkpoint

    // 开启checkpoint,参数为生成检查点的时间间隔,默认500ms(watermark默认为200ms)
    // 还可以传第二个参数CheckpointingMode:EXACTLY_ONCE/AT_LEAST_ONCE
    env.enableCheckpointing(1000L)

    // checkpoint的配置
    val chkpConfig = env.getCheckpointConfig
    // 设置生成检查点的时间间隔(前一次Checkpoint开始到下一次Checkpoint开始的时间)
    chkpConfig.setCheckpointInterval(1000L)
    // 设置CheckpointingMode:EXACTLY_ONCE/AT_LEAST_ONCE
    chkpConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 所有任务完成一个checkpoint的超时时间,一般给分钟级别
    chkpConfig.setCheckpointTimeout(60000L)
    // 同时进行的checkpoint数量,默认1,(一个checkpoint指的是barrier从头到尾完成所有算子的状态保存)
    chkpConfig.setMaxConcurrentCheckpoints(2)
    // 前一次Checkpoint结束到下一次Checkpoint开始至少500ms
    /* 假设第一次Checkpoint在20:00:00开始,两次生成检查点的时间间隔为1s(1000ms)
       1.当Checkpoint处理时间为600ms时,原本下次Checkpoint在20:00:01开始,
       因为最小间隔为500ms,所以现在需要延后100ms,600+500=1.1s
       2.当Checkpoint处理时间为400ms时,下次Checkpoint在20:00:01开始,
       因为最小间隔为500ms,400+500=900ms,小于1s,由于不能提前,所以还是20:00:01开始下次Checkpoint
     */
    // setMaxConcurrentCheckpoints(2)与setMinPauseBetweenCheckpoints(500L)冲突,至少间隔500ms就不可能同时发生了
    chkpConfig.setMinPauseBetweenCheckpoints(500L)
    // 是否使用Checkpoint做故障恢复,不管save point
    chkpConfig.setPreferCheckpointForRecovery(true)
    // 能容忍的Checkpoint失败次数
    chkpConfig.setTolerableCheckpointFailureNumber(3)
  • 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

4、保存点(save points)

Flink 还提供了可以自定义的镜像保存功能,就是保存点(savepoints)。

  • 原则上,创建保存点使用的算法与检查点完全相同,因此保存点可以认为就是具有一些额外元数据的检查点。
  • Flink不会自动创建保存点,因此用户(或者外部调度程序)必须明确地触发创建操作。
  • 保存点是一个强大的功能。除了故障恢复外,保存点可以用于:有计划的手动备份,更新应用程序,版本迁移,暂停和重启应用,等等。

5、重启策略

// 固定时间间隔重启,参数为重启次数,重启时间间隔
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))
  • 1
  • 2
// 参数为重启次数,所有重启使用的时间,两次重启时间间隔
env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.minutes(5), Time.seconds(10)))
  • 1
  • 2
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/正经夜光杯/article/detail/876335
推荐阅读
相关标签
  

闽ICP备14008679号