修改State Backend的两种方式
第一种：单任务调整
修改当前任务代码
env.setStateBackend(new FsStateBackend("hdfs://namenode:9000/flink/checkpoints"));
或者new MemoryStateBackend()
或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】
第二种：全局调整
修改flink-conf.yaml
state.backend: filesystem
state.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints
注意：state.backend的值可以是下面几种：jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)
Checkpoint的高级选项
默认checkpoint功能是disabled的，想要使用的时候需要先启用checkpoint开启之后，默认的checkPointMode是Exactly-once
//配置一秒钟开启一个checkpoint env.enableCheckpointing(1000) //指定checkpoint的执行模式 //两种可选： //CheckpointingMode.EXACTLY_ONCE：默认值 //CheckpointingMode.AT_LEAST_ONCE env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) 一般情况下选择CheckpointingMode.EXACTLY_ONCE，除非场景要求极低的延迟（几毫秒） 注意：如果需要保证EXACTLY_ONCE，source和sink要求必须同时保证EXACTLY_ONCE
//如果程序被cancle，保留以前做的checkpoint env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) 默认情况下，检查点不被保留，仅用于在故障中恢复作业，可以启用外部持久化检查点，同时指定保留策略: ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:在作业取消时保留检查点，注意，在这种情况下，您必须在取消后手动清理检查点状态 ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION：当作业在被cancel时，删除检查点，检查点仅在作业失败时可用
//设置checkpoint超时时间 env.getCheckpointConfig.setCheckpointTimeout(60000) //Checkpointing的超时时间，超时时间内没有完成则被终止
//Checkpointing最小时间间隔，用于指定上一个checkpoint完成之后 //最小等多久可以触发另一个checkpoint，当指定这个参数时，maxConcurrentCheckpoints的值为1 env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
//设置同一个时间是否可以有多个checkpoint执行 env.getCheckpointConfig.setMaxConcurrentCheckpoints(1) 指定运行中的checkpoint最多可以有多少个 env.getCheckpointConfig.setFailOnCheckpointingErrors(true) 用于指定在checkpoint发生异常的时候，是否应该fail该task，默认是true，如果设置为false，则task会拒绝checkpoint然后继续运行
Flink的重启策略
Flink支持不同的重启策略，这些重启策略控制着job失败后如何重启。集群可以通过默认的重启策略来重启，这个默认的重启策略通常在未指定重启策略的情况下使用，而如果Job提交的时候指定了重启策略，这个重启策略就会覆盖掉集群的默认重启策略。
概览
默认的重启策略是通过Flink的 flink-conf.yaml 来指定的，这个配置参数 restart-strategy 定义了哪种策略会被采用。如果checkpoint未启动，就会采用 no restart 策略，如果启动了checkpoint机制，但是未指定重启策略的话，就会采用 fixed-delay 策略，重试 Integer.MAX_VALUE 次。请参考下面的可用重启策略来了解哪些值是支持的。
每个重启策略都有自己的参数来控制它的行为，这些值也可以在配置文件中设置，每个重启策略的描述都包含着各自的配置值信息。
| 重启策略 | 重启策略值 |
| Fixed delay | fixed-delay |
| Failure rate | failure-rate |
| No restart | None |
除了定义一个默认的重启策略之外，你还可以为每一个Job指定它自己的重启策略，这个重启策略可以在 ExecutionEnvironment 中调用 setRestartStrategy() 方法来程序化地调用，注意这种方式同样适用于 StreamExecutionEnvironment。
下面的例子展示了如何为Job设置一个固定延迟重启策略，一旦有失败，系统就会尝试每10秒重启一次，重启3次。
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, // 重启次数 Time.of(10, TimeUnit.SECONDS) // 延迟时间间隔 ))
固定延迟重启策略(Fixed Delay Restart Strategy)
固定延迟重启策略会尝试一个给定的次数来重启Job，如果超过了最大的重启次数，Job最终将失败。在连续的两次重启尝试之间，重启策略会等待一个固定的时间。
重启策略可以配置flink-conf.yaml的下面配置参数来启用，作为默认的重启策略:
restart-strategy: fixed-delay
| 配置参数 | 描述 | 默认值 |
| restart-strategy.fixed-delay.attempts | 在Job最终宣告失败之前，Flink尝试执行的次数 | 1，如果启用checkpoint的话是Integer.MAX_VALUE |
| restart-strategy.fixed-delay.delay | 延迟重启意味着一个执行失败之后，并不会立即重启，而是要等待一段时间。 | akka.ask.timeout,如果启用checkpoint的话是1s |
例子:
restart-strategy.fixed-delay.attempts: 3 restart-strategy.fixed-delay.delay: 10 s
固定延迟重启也可以在程序中设置:
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, // 重启次数 Time.of(10, TimeUnit.SECONDS) // 重启时间间隔 ))
失败率重启策略
失败率重启策略在Job失败后会重启，但是超过失败率后，Job会最终被认定失败。在两个连续的重启尝试之间，重启策略会等待一个固定的时间。
失败率重启策略可以在flink-conf.yaml中设置下面的配置参数来启用:
restart-strategy:failure-rate
| 配置参数 | 描述 | 默认值 |
| restart-strategy.failure-rate.max-failures-per-interval | 在一个Job认定为失败之前，最大的重启次数 | 1 |
| restart-strategy.failure-rate.failure-rate-interval | 计算失败率的时间间隔 | 1分钟 |
| restart-strategy.failure-rate.delay | 两次连续重启尝试之间的时间间隔 | akka.ask.timeout |
例子:
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
失败率重启策略也可以在程序中设置:
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, // 每个测量时间间隔最大失败次数 Time.of(5, TimeUnit.MINUTES), //失败率测量的时间间隔 Time.of(10, TimeUnit.SECONDS) // 两次连续重启尝试的时间间隔 ))
无重启策略
Job直接失败，不会尝试进行重启
restart-strategy: none
无重启策略也可以在程序中设置
val env = ExecutionEnvironment.getExecutionEnvironment() env.setRestartStrategy(RestartStrategies.noRestart())
