案例

socket 服务器向 Flink 程序发送数据，格式如下：

1,30

解释：

1：表示传感器
30：表示发送的温度

打印各个传感器的 温度最大值

当输入数据格式错误时，程序崩溃，3秒后会重新启动，从checkpoint恢复后，继续运行，连续错误2次后，程序退出

java代码

package checkpoint;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
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.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class S2报错保存socket {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 必设置，每隔 1000 ms 执行一次checkpoint。
        env.enableCheckpointing(1000);//一般秒级、分钟级
//设置保存文件路径，生产环境中应该是 HDFS
        env.setStateBackend(new FsStateBackend("file:///D:/flink_cp"));
//env.setStateBackend(new RocksDBStateBackend("file:///D:/bigdata/flink_java/state/RocksDB",true));
// 设置2个checkpoint之间最少等待时间，这里至少要等到500ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//checkpoint容忍的错误数量，达到此数量，就让该任务失败，默认值是：0。这里容忍3个错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
//取消任务时，是否清理checkpoint
// true：取消任务时，会删除外部checkpoint，默认值
// false：取消任务时，保留外部checkpoint
//这里设置不删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//设置checkpoint的执行模式：EXACTLY_ONCE，默认
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 设置checkpoint的超时时间，如果在60s内尚未完成说明本次checkpoint失败，则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);//这里设置10分钟
// 设置同一时间有多少个checkpoint可以同时执行。与 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); 可能冲突
//env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//默认为1

// 重启策略：程序出现异常时，重启2次，每次延迟3秒中重启，超过2次退出程序
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2,3000));

        DataStreamSource<String> dss = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<Tuple2<String,Double>> soso = dss
                .map(new MapFunction<String,Tuple2<String,Double>>(){

                        @Override
                        public Tuple2<String,Double> map(String s) throws Exception {
                            String[] arr = s.split(",");
                            double temp = Double.parseDouble(arr[1]);
                            return Tuple2.of(arr[0],temp);
                        }
                    })
                .keyBy(new KeySelector<Tuple2<String,Double>,String>(){

                    @Override
                    public String getKey(Tuple2<String, Double> value) throws Exception {
                        return value.f0;
                    }
                })
                .maxBy(1);
        soso.print();

        env.execute();
    }


}

启动

启动 socket 测试软件

启动 flink 程序

第一次输入

在 socket 测试软件中输入：

1,20

在 flink控制台中打印：

1,20

因为只有一条数据，所以也是最大值

第二次输入

在 socket 测试软件中输入：

1,25

在 flink控制台中打印：

1,25

因为该条记录的温度最高

第三次输入

在 socket 测试软件中输入：

flink 程序报错，3秒后，再次运行

第四次输入

在 socket 测试软件中输入：

1,10

在 flink控制台中打印：

1,25

是上一次的温度最高值，说明，程序崩溃后，根据 checkpoint 恢复历史数据了

观察

D:\flink_cp\6088d78ac90eb147309d1f9ceb7d31a7

目录下的

chk-xxx

文件，每间隔 1s 就重新创建一个，其编号一直增长

原文出处：/show_1IX22DcnUOhE.html