（3）Flink on k8s checkpoint 状态写入pv
1.简介
大家都知道，Flink 是一个有状态的分布式流式计算引擎，flink 中的每个function或者是operator都可以是有状态的，为了使得状态可以容错，flink引入了checkpoint机制。checkpoint使得flink能够恢复作业的状态和位置，从而为作业提供与无故障执行相同的语义。
CheckPoint的触发和状态数据管理主要由JobManager负责，JobManager需要将checkpoint生成的状态信息文件保存在外部Storage中，在基于On Yarn的部署模式下，这个Storage通常是HDFS，在On Kubernetes的模式下，这个Storage通常是PV，这样可以使Flink On K8s减少对其他第三方组件的依赖，比如Hadoopp的依赖。
2.Checkpoint 程序开发
为了演示在K8S 上运行flink checkpoint程序，新开发了一个checkpoint演示程序，这个程序实现的是WordCount单词统计功能。
在这个示例中，上游的Source使用Kafka，由Kafka的生产者程序往对应的topic发送字符流信息，在实际生成环境中，Flink的Source通常也是Kafka。
Flink Checkpoint程序读取Kafka Topic的字符流，然后进行分词和单词统计，并将统计结果Sink输出到MySQL数据库中。
为了确保即使程序终止重启后，可以在原有的单词统计结果上累计单词的统计结果，例如在Flink程序重启前，我们统计Hello的单词出现了5次，在重启程序继续消费Kafka的字符流数据时，当首次遇到包含Hello单词的字符流时，此时Hello的统计结果是6，而不是1。为此，我们在程序里开启checkpoint功能，这样JobManager就能定期周期性的触发checkpoint操作，将当前的统计结果，也就是checkpoint状态信息保存起来。
1、flink 主程序
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;
import java.util.Arrays;
import java.util.Properties;
public class StreamWordCountWithCP {
private static Logger logger = Logger.getLogger(StreamWordCountWithCP.class);
public static void main(String[] args) throws Exception {
logger.info("******************* StreamWordCountWithCP job start *******************");
// kafka server
String kafkaServer = "172.16.252.129:9092,172.16.252.130:9092,172.16.252.131:9092";
// kafka topic
String kafkaTopic = "flink_test";
// mysql数据库ip
String dbHost = "172.16.252.113";
// mysql数据库端口
String dbPort = "3306";
// 数据库名称
String dbName = "flink_test";
// 结果表
String table = "wc";
// checkpoint文件保存路径
String checkpointPath = "file:///Users/willwang/flink-on-k8s-demo/checkpoint";
// checkpoint保存时间间隔，默认10s
long interval = 10000;
// 并行度
int parallelism = 1;
// 从程序传参中获取参数
if (args != null && args.length == 9) {
kafkaServer = args[0];
kafkaTopic = args[1];
dbHost = args[2];
dbPort = args[3];
dbName = args[4];
table = args[5];
checkpointPath = args[6];
interval = Long.parseLong(args[7]);
parallelism = Integer.parseInt(args[8]);
logger.info("******************* kafkaServer=" + args[0] + ", " +
"kafkaTopic=" + args[1] + ", " +
"dbHost=" + args[2] + ", " +
"dbPort=" + args[3] + ", " +
"dbName=" + args[4] + ", " +
"table=" + args[5] + ", " +
"checkpointPath=" + args[6] + ", " +
"interval=" + args[7] + ", " +
"parallelism=" + args[8]);
}
// 0. 创建流式执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(parallelism);
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
// 1. 配置checkpoint
env.enableCheckpointing(interval);
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
checkpointConfig.setMaxConcurrentCheckpoints(1);
checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
checkpointConfig.enableUnalignedCheckpoints();
checkpointConfig.setCheckpointStorage(checkpointPath);
// 2. 配置Kafka
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", kafkaServer);
properties.setProperty("group.id", "wc-consumer-group");
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("auto.offset.reset", "latest");
// 3. 设置kafka source
DataStreamSource stream = env.addSource(new FlinkKafkaConsumer(
kafkaTopic,
new SimpleStringSchema(),
properties
));
// 4. 转换数据格式
SingleOutputStreamOperator wordAndOne = stream
.flatMap((String line, Collector words) -> {
Arrays.stream(line.split(" ")).forEach(words::collect);
})
.returns(Types.STRING)
.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));
// 5. 分组
KeyedStream wordAndOneKS = wordAndOne
.keyBy(t -> t.f0);
// 6. 求和
SingleOutputStreamOperator result = wordAndOneKS
.sum(1);
// 7. 设置自定义sink，结果输出到MySQL
result.addSink(new MySQLSink(dbHost, dbPort, dbName, table));
env.execute("StreamWordCountWithCP");
}
}
2、sink mysql工具类
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.log4j.Logger;
import java.sql.*;
public class MySQLSink extends RichSinkFunction {
private static Logger logger = Logger.getLogger(MySQLSink.class);
private Connection conn;
private PreparedStatement insertPS;
private PreparedStatement updatePS;
private String dbHost;
private String dbPort;
private String dbName;
private String table;
public MySQLSink(String dbHost, String dbPort, String dbName, String table) {
this.dbHost = dbHost;
this.dbPort = dbPort;
this.dbName = dbName;
this.table = table;
}
@Override
public void open(Configuration parameters) throws Exception {
String url = "jdbc:mysql://"+dbHost+":"+dbPort+"/" + dbName + "?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC";
//建立mysql连接
conn = DriverManager.getConnection(url, "root", "111...aaa");
String sql = "INSERT INTO "+table+" (`word`, `cnt`) VALUES(?,?);";
insertPS = conn.prepareStatement(sql);
String sql2 = "UPDATE "+table+" set cnt = ? where word = ?;";
updatePS = conn.prepareStatement(sql2);
System.out.println("自定义sink，open数据库链接 =====");
logger.info("*** 自定义sink，open数据库链接 =====");
}
@Override
public void close() throws Exception {
if(insertPS != null){
insertPS.close();
}
if(updatePS != null){
updatePS.close();
}
if(conn != null){
conn.close();
}
System.out.println("自定义sink，close数据库连接 =====");
logger.info("*** 自定义sink，close数据库连接 =====");
}
@Override
public void invoke(Tuple2 value, Context context) throws Exception {
String sql = "select count(*) from "+table+" where word = '" + value.f0 + "'";
System.out.println(sql);
logger.info("*** " + sql);
Statement statement = conn.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
long cnt = 0;
while (resultSet.next()) {
cnt = resultSet.getLong(1);
}
resultSet.close();
statement.close();
if (cnt > 0) {
System.out.println("update value=" + value);
updatePS.setLong(1,value.f1);
updatePS.setString(2,value.f0);
updatePS.executeUpdate();
} else {
System.out.println("insert value=" + value);
insertPS.setString(1,value.f0);
insertPS.setLong(2,value.f1);
insertPS.executeUpdate();
}
}
}
3.K8S 测试配置
1、创建checkpoint pvc 存储checkpoint的数据
[root@k8s-demo001 ~]# cat flink-checkpoint-application-pvc.yaml
# Flink checkpoint 持久化存储pvc
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: flink-checkpoint-application-pvc # checkpoint pvc名称
namespace: flink # 指定归属的名命空间
spec:
storageClassName: nfs-storage #sc名称，更改为实际的sc名称
accessModes:
- ReadWriteMany #采用ReadWriteMany的访问模式
resources:
requests:
storage: 1Gi #存储容量，根据实际需要更改
2、通过application 模式进行测试
[root@k8s-demo001 ~]# cat application-deployment-checkpoint.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
namespace: flink
name: application-deployment-checkpoint # flink 集群名称
spec:
image: flink:1.13.6 # flink基础镜像
flinkVersion: v1_13 # flink版本，选择1.13
imagePullPolicy: IfNotPresent # 镜像拉取策略，本地没有则从仓库拉取
ingress: # ingress配置，用于访问flink web页面
template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
annotations:
nginx.ingress.kubernetes.io/rewrite-target: "/$2"
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.checkpoints.dir: file:///opt/flink/checkpoints
serviceAccount: flink
jobManager:
resource:
memory: "1024m"
cpu: 1
taskManager:
resource:
memory: "1024m"
cpu: 1
podTemplate:
spec:
containers:
- name: flink-main-container
env:
- name: TZ
value: Asia/Shanghai
volumeMounts:
- name: flink-jar # 挂载nfs上的jar
mountPath: /opt/flink/jar
- name: flink-checkpoints # 挂载checkpoint pvc
mountPath: /opt/flink/checkpoints
volumes:
- name: flink-jar
persistentVolumeClaim:
claimName: flink-jar-pvc
- name: flink-checkpoints
persistentVolumeClaim:
claimName: flink-checkpoint-application-pvc
job:
jarURI: local:///opt/flink/jar/flink-on-k8s-demo-1.0-SNAPSHOT-jar-with-dependencies.jar # 使用pv方式挂载jar包
entryClass: org.fblinux.StreamWordCountWithCP
args: # 传递到作业main方法的参数
- "172.16.252.129:9092,172.16.252.130:9092,172.16.252.131:9092"
- "flink_test"
- "172.16.252.113"
- "3306"
- "flink_test"
- "wc"
- "file:///opt/flink/checkpoints"
- "10000"
- "1"
parallelism: 1
upgradeMode: stateless
3、访问flink ui：验证checkpoint 也可以落地到实际的路径
http://flink.k8s.io/flink/application-deployment-checkpoint/#/job/453b1bd5f90756e7afe929c3e71f747b/checkpoints
4、通过往Kafka 发送数据，flink 程序也可以写入到mysql
对于application 作业，要是终止和重启这个作业，那么整个application的集群也会一起关闭掉，jobmanager和taskmanager也会一并释放掉，checkpoint数据也会丢失，因为新的作业checkpoint路径是会变化的，这个时候需要查询pv获取最后一次的checkpoint文件路径。
1、终止作业
kubectl delete -f application-deployment-checkpoint.yaml
2、编写重启作业的yaml
前面的内容和第一次提交checkpoint是一样的，只是在第一次启动需要指定initialSavepointPath，只有这点不一样。
[root@k8s-demo001 ~]# cat application-deployment-checkpoint.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
namespace: flink
name: application-deployment-checkpoint # flink 集群名称
spec:
image: flink:1.13.6 # flink基础镜像
flinkVersion: v1_13 # flink版本，选择1.13
imagePullPolicy: IfNotPresent # 镜像拉取策略，本地没有则从仓库拉取
ingress: # ingress配置，用于访问flink web页面
template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
annotations:
nginx.ingress.kubernetes.io/rewrite-target: "/$2"
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.checkpoints.dir: file:///opt/flink/checkpoints
serviceAccount: flink
jobManager:
resource:
memory: "1024m"
cpu: 1
taskManager:
resource:
memory: "1024m"
cpu: 1
podTemplate:
spec:
hostAliases:
- ip: "172.16.252.129"
hostnames:
- "kafka-01"
- ip: "172.16.252.130"
hostnames:
- "kafka-02"
- ip: "172.16.252.131"
hostnames:
- "kafka-03"
containers:
- name: flink-main-container
env:
- name: TZ
value: Asia/Shanghai
volumeMounts:
- name: flink-jar # 挂载nfs上的jar
mountPath: /opt/flink/jar
- name: flink-checkpoints # 挂载checkpoint pvc
mountPath: /opt/flink/checkpoints
volumes:
- name: flink-jar
persistentVolumeClaim:
claimName: flink-jar-pvc
- name: flink-checkpoints
persistentVolumeClaim:
claimName: flink-checkpoint-application-pvc
job:
jarURI: local:///opt/flink/jar/flink-on-k8s-demo-1.0-SNAPSHOT-jar-with-dependencies.jar # 使用pv方式挂载jar包
entryClass: org.fblinux.StreamWordCountWithCP
args: # 传递到作业main方法的参数
- "172.16.252.129:9092,172.16.252.130:9092,172.16.252.131:9092"
- "flink_test"
- "172.16.252.113"
- "3306"
- "flink_test"
- "wc"
- "file:///opt/flink/checkpoints"
- "10000"
- "1"
initialSavepointPath: /opt/flink/checkpoints/4a81f02e391f129fa647121e4014226f/chk-70/ # checkpoint文件绝对路径
parallelism: 1
upgradeMode: stateless
重新提交，在通过Kafka写入数据，观察mysql表就可以看到是从checkpoint恢复的数据
