Flink状态管理与恢复

Flink支持不同的重启策略，以便在故障发生时（任务失败时）控制作业如何重启，当然如果你在没有学checkpoint的时候，你可能会问：为什么我的flink的任务在任务失败的时候
1

直接挂了，没有重启呢？是的，如果flink任务没有启用checkpoint，则使用无重启策略，那么当你配置了checkpoint的时候，重启策略又是怎么样的呢？这便是我们

本篇博客的主要内容：Restart Strategies(flink的重启策略)

常用的flink重启策略
Flink支持不同的重启策略，以在故障发生时控制作业如何重启，集群在启动时会伴随一个默认的重启策略，在没有定义具体重启策略时会使用该默认策略。 如果在工作提交时指定了一个重启策略，

该策略会覆盖集群的默认策略，默认的重启策略可以通过Flink的配置文件flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。

常用的重启策略有3种：

固定间隔 (Fixed delay)、失败率 (Failure rate)、无重启 (No restart)。

需要说明的是:

如果没有启用 checkpoint，默认使用无重启 (no restart) 策略；

如果启用了 checkpoint，但没有配置重启策略，则默认使用固定间隔 (fixed-delay) 策略，其中 Integer.MAX_VALUE参数是尝试重启次数（不知道你发现没？这就是无数次重启啊！）

flink重启策略的步骤
重启策略可以在flink-conf.yaml中配置，表示全局的配置。也可以在应用代码中动态指定，会覆盖全局配置。

重启策略之固定间隔 (Fixed delay)

1. 单任务调整（任务发生故障时重启xxx次,每次间隔是xxx秒.）

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
// 尝试重启的次数
Time.of(10, TimeUnit.SECONDS) // 间隔
));
2. 全局调整:修改flink-conf.yaml

restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10 s
重启策略之失败率 (Failure rate)

1. 单任务调整

env.setRestartStrategy(RestartStrategies.failureRateRestart(
// 一个时间段内的最大失败次数
Time.of(5, TimeUnit.MINUTES), // 衡量失败次数的是时间段
Time.of(10, TimeUnit.SECONDS) // 间隔
));
2. 全局调整:修改flink-conf.yaml

restart-strategy: failure-rate
restart-strategy.failure-rate.max-failures-per-interval: 3
restart-strategy.failure-rate.failure-rate-interval: 5 min
restart-strategy.failure-rate.delay: 10 s
重启策略之无重启 (No restart)

1. 单任务调整

env.setRestartStrategy(RestartStrategies.noRestart());
2. 全局调整:修改flink-conf.yaml

restart-strategy: none
重启策略实战操作　
上面介绍了这么多，接下来我们就实际操作一下，其实非常简答：

代码示例：

package Stream_example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**

• @ClassName SocketWindowWordCountJavaCheckPoint

• {功能描述:Flink当中的checkpoint机制

• 滑动窗口计算:实现每隔2秒对最近1个小时的数据进行汇总计算}

• @Author Admin

• CREATE 2020/10/5 18:39

• @Version 1.0.0
  */
  public class SocketWindowWordCountJavaCheckPoint {
  public static void main(String[] args) throws Exception {
  //获取Flink的运行环境.
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

   //开启flink的checkpoint功能:每隔1000 ms启动一个检查点（设置checkpoint的生命周期.）
   env.enableCheckpointing(1000);
  
   //checkpoint高级选项设置.
   //设置checkpoint的模式为exactly-once（这也是默认值）
   env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
   //确保检查点之间至少有500ms的间隔（即checkpoint的最小间隔）
   env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
   //检查点必须在1min之内完成,否则会被丢弃（checkpoint的超时时间）
   env.getCheckpointConfig().setCheckpointTimeout(60000);
   //同一时间只允许操作一个检查点
   env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
   //Flink程序即使被cancel后,也会保留checkpoint数据,以便根据实际需要恢复到指定的checkpoint.
   env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  
   //设置statebackend,指定state和checkpoint的数据存储位置（checkpoint的数据必须得有一个可以持久化存储的地方）
   env.setStateBackend(new FsStateBackend("hdfs://s101:9000/flink/checkpoints"));
  
   //重启策略采用固定间隔 (Fixed delay) :任务发生故障时重启3次,每次间隔是10秒.
   env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,10000));
  
   //连接Socket获取输入的数据.
   DataStreamSource  socketTextStream = env.socketTextStream("192.168.140.103", 8888, "\n");
  
   DataStream> tuple2SingleOutputStreamOperator = socketTextStream.flatMap(new FlatMapFunction>() {
       /**
        * @param value socket传过来的一行一行的数据.
        * @param out
        * @throws Exception
        */
       @Override
       public void flatMap(String line, Collector> out) throws Exception {
           String[] splited = line.split("\\W+");
           for (String word : splited) {
               out.collect(new Tuple2<>(word, 1));
           }
       }
   });
  
   //指定时间窗口大小为1*60*60s,指定时间间隔为2s.
   DataStream> sumData = tuple2SingleOutputStreamOperator.keyBy(0).timeWindow(Time.seconds(3600), Time.seconds(2)).sum(1);
   sumData.print();
  
   env.execute("SocketWindowWordCountJavaCheckPoint");
  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

  }
  }
  当我们运行代码的时候，为了让代码出现问题，我们不启动nc服务：

Stream_example.SocketWindowWordCountJavaCheckPoint FlinkExample-1.0-SNAPSHOT-jar-with-dependencies.jar

在重启3次之后，任务失败结束：

失败日志我们也是可以看到的：

ok，flink的失败重启策略我们就讲完了，是不是很简单！

后续我在用flink1.9进行测试的时候，没有看到重启的Restart的标志，但是我们可以从下面的标志看到重启的痕迹.