在flink里，算法基本上带有状态的，而在我们处理元素或事件时状态会存储数据。为了让大烽据框架的状态有容错功能，flink提出checkpointing机制。
一、checkpointing 与 flink容错机制
当错误或异常发生时，checkpointing允许flink去恢复状态以及流所在的位置。
Flink's checkpointing实现机制是将数据流与状态永久存储在某一个地方，因此让flink拥有容错机制，需要有两个前置条件：
- 1.必须拥有持久化或可靠的数据源，该数据源可以在一定的时间范围内提供回放记录的功能(replay records for certian of time)；比例：kafka、rabbitmq、amazon kinese、google pubsub 或 分布式文件系统 hdfs、s3、gfs、nfs、ceph...等
- 2.状态的持久化存储，典型的是使用分布式文件存储；比如hdfs、s3、gfs、nfs、ceph...等
如何开启及配置checkpointing功能
默认情况下，flink在checkpointing是disabled(禁用)的，开启方法：enableCheckpointing(n)) on StreamExecutionEnvironment。
n代表时间
checkpointing相关参数
- exactly-once vs at-least-once：精准消费一次/至少消费一次。配置方法是通过
enableCheckpointing(n)相关操作选择两种模式之一。在大部份情况下，exactly-once是比较好的一个选择。 at-least-once用在对于低延迟只有的环情况(比如只可延迟几毫秒)。 - checkpoint timeout: checkpoint过程当中所消耗时间的设置，保存不超时或超时中断
- minimum time between checkpoints：最小时间，在两个checkpointing之间。如果该值设定为5000，而下一个checkpoint一般会上一个checkpoint结束后5秒开始。
- number of cocurrent checkpoints：并发checkpoint的数量。默认只要有一个checkpoint进程正在运行另外一个不会触发。
- externalized checkpoints：可配置一个周期性的外部持久化的checkpoints。通过该功能可以将元数据信息等写到外部的存储中，此种方式下当job运行过程中出现异常，flink不会自动删除这些外部存储的checkpoint数据或信息。该模式中以让flink出现异常或操作中断时从外部存储的数据点中恢复数据。
- fail/continue task on checkpoint erros: 当失败发生时，是继续执行还是中断操作。默认为中断执行。
- prefer checkpoint for recovery：优先从checkpoints点中恢复数据
- unaliged checkpoints:
代码
package com.flink.state;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class CheckpointApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
env.enableCheckpointing(5000);
DataStreamSource<String> source = env.socketTextStream("localhost",9527);
source.print();
env.execute("CheckPointApp");
}
}
checkpoint重启策略
重启策略与故障恢复策略，来控制任务的重启。重启策略决定了(是否或什么时候）以什么方式重启。故障恢复策略决定任务恢复的点。
策略清单
image.png
设定重启策略
设定重启策略代码
重启策略片段
image.png
详情看官网
重启策略举例
package com.flink.state;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.util.Locale;
import java.util.concurrent.TimeUnit;
public class CheckpointApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
/**
* 不开启checkpoint：不重启
* 开启了checkpoint：
* 1) 没有配置重启策略: Integer。MAX_VALUE
* 2) 如果配置了重启策略，就使用我们配置的重启策略覆盖默认的
*
* 重启策略的配置
* 1) code
* 2) yaml
*/
env.enableCheckpointing(5000);
env.setRestartStrategy(
//偿试3次，间隔5秒。
RestartStrategies.fixedDelayRestart(3, Time.of(5, TimeUnit.SECONDS))
);
DataStreamSource<String> source = env.socketTextStream("localhost",9527);
source.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
if (value.contains("pk")){
throw new RuntimeException("PK哥来了，快跑..!");
}else{
return value.toLowerCase(Locale.ROOT);
}
}
}).flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
String[] splits = value.split(",");
for (String split : splits) {
out.collect(split);
}
}
}).map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return Tuple2.of(value,1);
}
}).keyBy(x -> x.f0).sum(1).print();
//source.print();
env.execute("CheckPointApp");
}
}
ExternalizedCheckpoints
当作业失败或人工取消息时，是否可留之前的checkpoint记录
//当job取消以后，是否清楚外部的checkpoint记录
//是否保留
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
网友评论