- ``: 检查点之间的时间间隔（以毫秒为单位）。在此间隔内，系统将生成新的检查点
SET = 6000;
- `-failed-checkpoints`: 允许的连续失败检查点的最大数量。如果连续失败的检查点数量超过此值，作业将失败。
SET -failed-checkpoints = 10;
- ``: 检查点的超时时间（以毫秒为单位）。如果在此时间内未完成检查点操作，作业将失败。
SET =600000;
- `-checkpoint-retention`: 外部化检查点的保留策略。`RETAIN_ON_CANCELLATION`表示在作业取消时保留外部化检查点。
SET -checkpoint-retention = RETAIN_ON_CANCELLATION;
- ``: 检查点模式。`EXACTLY_ONCE`表示每个检查点只会在作业处理完全一次时生成。
SET = EXACTLY_ONCE;
- ``: 检查点是否对齐。如果设置为`true`，则检查点将在作业的所有任务完成之前生成。
SET = true;
- `-concurrent-checkpoints`: 并发生成检查点的最大数量。在此数量的检查点生成之前，不会生成新的检查点。
SET -concurrent-checkpoints = 1;
- `-retained`: 保留的检查点数量。超过此数量的检查点将被删除
SET -retained = 3;
暂未使用
SET = 6000;
SET -failed-checkpoints = 10;
SET =600000;
SET -checkpoint-retention = RETAIN_ON_CANCELLATION;
SET = EXACTLY_ONCE;
SET = true;
SET -concurrent-checkpoints = 1;
SET -retained = 3;
yaml 文件中 起作用的配置信息如下:
flinkConfiguration:
: "36"
: rocksdb
-storage: filesystem
-retained: "3"
: "true"
: file:///flink-data/savepoints
: file:///flink-data/checkpoints
: kubernetes
high-availability: # JobManager HA
: file:///opt/flink/flink_recovery # JobManager HA数据保存路径
serviceAccount: flink
登录到
kubectl -n flink exec -it session-deployment-only-taskmanager-2-1 bash
查看 cat
: 6124
: /generation:2
-retained: 3
: 2
: kubernetes
-id: session-deployment-only
: file:///flink-data/savepoints
: 4.0
-account: flink
-id: session-deployment-only
-storage: filesystem
: file:///opt/flink/flink_recovery
: 1
: localhost:5000/flink-sql:1.14.21
: 1
: flink
: 36
: ClusterIP
: 6123
: blockOwnerDeletion:false,apiVersion:/v1beta1,kind:FlinkDeployment,uid:fadef756-d327-4f19-b1b4-181d92c659eb,name:session-deployment-only,controller:false
: 6048m
:
-template-file: /tmp/flink_op_generated_podTemplate_4716243666145995447.yaml
: true
: false
: kubernetes-session
: 1024m
: 6122
-policy: IfNotPresent
-mode: NORMAL
high-availability:
: 1.0
: rocksdb
$: v1_14
: file:///flink-data/checkpoints
只产生3个chk文件了
一个chk-X代表了一次Checkpoint信息，里面存储Checkpoint的元数据和数据。
taskowned: TaskManagers拥有的状态
shared: 共享的状态
默认情况下，如果设置了Checkpoint选项，Flink只保留最近成功生成的1个Checkpoint。当Flink程序失败时，可以从最近的这个Checkpoint来进行恢复。但是，如果我们希望保留多个Checkpoint，并能够根据实际需要选择其中一个进行恢复，这样会更加灵活。Flink支持保留多个Checkpoint，需要在Flink的配置文件conf/中，添加如下配置指定最多需要保存Checkpoint的个数，例如指定保留最近的10个Checkpoint(就是保留上面的10个chk-X)：
-retained: 10
ps1：Checkpoint目录如果删除，任务就无法指定从Checkpoint恢复了
ps2：如果job是失败了而不是手动cancel，那么无论选择上面哪种策略，state记录都会保留下来
ps3：使用RocksDB来作为增量checkpoint的存储，可以进行定期Lazy合并清除历史状态。
最后的yaml配置如下：
#Flink Session集群 源码请到
apiVersion: /v1beta1
kind: FlinkDeployment
metadata:
namespace: flink
name: session-deployment-only
spec:
image: 192.168.1.249:16443/bigdata/flink-sql:1.14.21
#image: localhost:5000/flink-sql:1.14.21
flinkVersion: v1_14
#imagePullPolicy: Never # 镜像拉取策略，本地没有则从仓库拉取
imagePullPolicy: IfNotPresent
ingress: # ingress配置，用于访问flink web页面
template: "flink."
className: "nginx"
annotations:
/rewrite-target: "/"
flinkConfiguration:
: "48"
: rocksdb
-storage: filesystem
-retained: "20"
: "true"
: file:///opt/flink/volume/flink-sp
: file:///opt/flink/volume/flink-cp
high-availability: # JobManager HA
: file:///opt/flink/volume/flink-ha # JobManager HA数据保存路径
serviceAccount: flink
jobManager:
replicas: 2
resource:
memory: "1024m"
cpu: 1
taskManager:
replicas: 1
resource:
memory: "6048m"
cpu: 4
podTemplate:
spec:
hostAliases:
- ip: "192.168.1.236"
hostnames:
- ""
- ip: "192.168.1.246"
hostnames:
- ""
containers:
- name: flink-main-container
env:
- name: TZ
value: Asia/Shanghai
volumeMounts:
- name: flink-volume #挂载checkpoint pvc
mountPath: /opt/flink/volume
volumes:
- name: flink-volume
persistentVolumeClaim:
claimName: flink-checkpoint-pvc
pvc：
#Flinnk checkpoint 持久化存储pvc
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: flink-checkpoint-pvc # checkpoint pvc名称
namespace: flink # 指定归属的名命空间
spec:
storageClassName: nfs-client #sc名称，更改为实际的sc名称
accessModes:
- ReadWriteMany #采用ReadWriteMany的访问模式
resources:
requests:
storage: 20Gi #存储容量，根据实际需要更改