案例
socket 服务器向 Flink 程序发送数据，当数据格式错误时，程序崩溃重启后，仍然能继续运算，状态不丢失
提示：此案例不能用kafka做演示，读取的错误格式数据，程序崩溃重启后，仍然会读取该错误格式数据
格式如下：
1,30
解释：
- 1：表示传感器
- 30：表示发送的温度
打印各个传感器的 温度最大值
当输入数据格式错误时，程序崩溃，3秒后会重新启动，从checkpoint恢复后，继续运行，连续错误2次后，程序退出
flink java代码
关键：
- 每隔
5000 ms执行一次 checkpoint
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);
setCheckpoint(env);
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 {
System.out.println("receive:"+s);
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("result:");
env.execute();
}
private static void setCheckpoint(StreamExecutionEnvironment env){
// 必设置，每隔 5000 ms 执行一次checkpoint。
env.enableCheckpointing(5000);//一般秒级、分钟级
//设置保存文件路径，生产环境中应该是 HDFS
// env.setStateBackend(new FsStateBackend("hdfs://hadoop1:8020/"+appName+"/ck/"));
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));
/*
如果保存到HDFS中，需要设置用户，否则调试时，一般会以 windows 用户操作HDFS，会报错没有权限
- 学习期间，设置用户为root（hadoop超级管理员）
- 企业开发中，给hadoop添加用户，并设置密码、权限，这里设置此用户
*/
System.setProperty("HADOOP_USER_NAME", "root");
}
}
java 测试类
当 flink 程序连接socket后，每间隔1秒发送一个数据
数据格式：
1,11
1,12
最后一条数据格式错误，故意模拟错误
代码
package top.malaoshi.socket;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
public class Server {
public static void main(String[] args) {
int port = 9999;
try {
ServerSocket ss = new ServerSocket(port);
System.out.println("server is running,port:"+port);
while(true) {
Socket client = ss.accept();
InetAddress inetAddress = client.getInetAddress();
String clientAddress = inetAddress.getHostAddress();
int clientPort = client.getPort();
System.out.println("new client:"+clientAddress+":"+clientPort);
SocketThread socketThread = new SocketThread(client);
socketThread.start();
}
} catch (IOException e) {
e.printStackTrace();
}
}
public static class SocketThread extends Thread {
private Socket client;
public SocketThread(Socket client) {
this.client = client;
}
public void run() {
try {
OutputStream is = client.getOutputStream();
InetAddress inetAddress = client.getInetAddress();
String clientAddress = inetAddress.getHostAddress();
int clientPort = client.getPort();
OutputStreamWriter osw = new OutputStreamWriter(is);
// int max = 4; // flink还没来得及执行 checkpoint，重启后，从11开始算
int max = 17;
for (int i = 1; i < max; i++) {
String data = "1," + (i + 10)+"\r\n";
osw.write(data);
osw.flush();
System.out.println("send:"+clientAddress+":"+clientPort+", data:"+data);
Thread.sleep(1000);
}
osw.write("1\r\n"); // 发送错误数据
osw.flush();
}catch (Exception e){
System.out.println(e);
}
}
};
}
第一次测试
启动 java 测试类
设置 max = 4，最后一条 错误数据 在第4秒发出
启动 flink 程序
启动后，socket 连接 java 测试类
结果
由于 checkpoint 每 5 秒执行一次，也就是说，还没有 执行 checkpoint ，程序就出错了，重启时，仍然重新开始算
第二次测试
启动 java 测试类
设置 max = 17，最后一条 错误数据 大约在第17秒发出
启动 flink 程序
启动后，socket 连接 java 测试类
结果
在第 17 秒时，发送错误数据，由于 checkpoint 每 5 秒执行一次，所有有几条记录还没来得及保存
flink 程序崩溃重启后，最近的几条数据 无法恢复
如下图：
观察
D:\flink_cp\6088d78ac90eb147309d1f9ceb7d31a7
目录下的
chk-xxx
文件，每间隔 5s 就重新创建一个，其 编号一直增长