Flink state 详解

系统管理员 2021-09-21 00:20 336阅读 0赞

### checkPoint简介 ###

*  为了保证state的容错性，Flink需要对state进行checkpoint。
 *  Checkpoint是Flink实现容错机制最核心的功能，它能够根据配置周期性地基于Stream中各个Operator/task的状态来生成快照，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常
 *  Flink的checkpoint机制可以与(stream和state)的持久化存储交互的前提：
    
     *  持久化的source，它需要支持在一定时间内重放事件。这种sources的典型例子是持久化的消息队列（比如Apache Kafka，RabbitMQ等）或文件系统（比如HDFS，S3，GFS等）
     *  用于state的持久化存储，例如分布式文件系统（比如HDFS，S3，GFS等）

### checkPoint配置 ###

*  默认checkpoint功能是disabled的，想要使用的时候需要先启用
 *  checkpoint开启之后，默认的checkPointMode是Exactly-once
 *  checkpoint的checkPointMode有两种，Exactly-once和At-least-once
 *  Exactly-once对于大多数应用来说是最合适的。At-least-once可能用在某些延迟超低的应用程序（始终延迟为几毫秒）
 *  默认checkpoint功能是disabled的，想要使用的时候需要先启用

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); 
    // 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】
    env.enableCheckpointing(1000); 
    // 高级选项：
    // 设置模式为exactly-once （这是默认值）
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); 
    // 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); 
    // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
    env.getCheckpointConfig().setCheckpointTimeout(60000); 
    // 同一时间只允许进行一个检查点
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); 
    // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
    env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); 
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
    ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint

### State Backend(状态的后端存储) ###

*  默认情况下，state会保存在taskmanager的内存中，checkpoint会存储在JobManager的内存中。
 *  state 的store和checkpoint的位置取决于State Backend的配置
    
     *  env.setStateBackend(…)
 *  一共有三种State Backend
    
     *  MemoryStateBackend
     *  FsStateBackend
     *  **RocksDBStateBackend**

*  MemoryStateBackend
    
     *  state数据保存在java堆内存中，执行checkpoint的时候，会把state的快照数据保存到jobmanager的内存中
     *  基于内存的Memory state backend在生产环境下不建议使用
 *  FsStateBackend
    
     *  state数据保存在taskmanager的内存中，执行checkpoint的时候，会把state的快照数据保存到配置的文件系统中
     *  可以使用hdfs等分布式文件系统

*  RocksDBStateBackend
    
     *  RocksDB跟上面的都略有不同，它会在本地文件系统中维护状态，state会直接写入本地rocksdb中。同时它需要配置一个远端的filesystem uri（一般是HDFS），在做checkpoint的时候，会把本地的数据直接复制到filesystem中。fail over的时候从filesystem中恢复到本地
     *  RocksDB克服了state受内存限制的缺点，同时又能够持久化到远端文件系统中，比较适合在生产中使用

### State Backend使用方式 ###

修改State Backend的两种方式

*  第一种：单任务调整
    
     *  修改当前任务代码
     *  env.setStateBackend(new FsStateBackend("hdfs://namenode:9000/flink/checkpoints"));
     *  或者new MemoryStateBackend()
     *  或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】
 *  第二种：全局调整
    
     *  修改flink-conf.yaml
    
        state.backend: filesystem 
        state.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints
    
     *  注意：state.backend的值可以是下面几种：
        
         *  jobmanager(MemoryStateBackend)
         *  filesystem(FsStateBackend)
         *  rocksdb(RocksDBStateBackend)

### State backend演示 ###

第一种：单任务调整

启动连接socket zzy:9001的程序

./bin/flink run -m yarn-cluster -yn 1 -yjm 1024 -ytm 1024 -c com.zzy.bigdata.flink.SocketWindowWordCountJavaCheckPoint zzy_flink_learn.jar --port 9001

![format_png][]

[iknow@data-5-63 flink-1.7.2]$ ./bin/flink run -m yarn-cluster -yn 1 -yjm 1024 -ytm 1024 -c com.zzy.bigdata.flink.SocketWindowWordCountJavaCheckPoint zzy_flink_learn.jar --port 9001
    2019-03-06 12:03:15,057 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-iknow.
    2019-03-06 12:03:15,057 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-iknow.
    2019-03-06 12:03:15,325 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
    2019-03-06 12:03:15,415 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
    2019-03-06 12:03:15,415 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
    2019-03-06 12:03:15,421 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument yn is deprecated in will be ignored.
    2019-03-06 12:03:15,421 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument yn is deprecated in will be ignored.
    2019-03-06 12:03:15,511 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
    2019-03-06 12:03:15,819 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/iknow/zhangzhiyong/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
    2019-03-06 12:03:16,386 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1551789318445_0004
    2019-03-06 12:03:16,412 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1551789318445_0004
    2019-03-06 12:03:16,412 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
    2019-03-06 12:03:16,414 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
    2019-03-06 12:03:19,940 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
    Starting execution of program

如果zzy上未开启9001端口,到jobManager的web ui上看到会报下面的错

![format_png 1][]

代码里设置了checkpoint

//获取flink的运行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //默认checkpoint功能是disabled的，想要使用的时候需要先启用;每隔10000ms进行启动一个检查点【设置checkpoint的周期】
    env.enableCheckpointing(10000);
    // 高级选项：
    // 设置模式为exactly-once （这是默认值）
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    // 确保检查点之间有至少500ms的间隔【checkpoint最小间隔】
    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
    // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
    env.getCheckpointConfig().setCheckpointTimeout(60000);
    // 同一时间只允许进行一个检查点
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
    // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
    env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
    //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
    //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
    
    
    //设置statebackend
    
    //env.setStateBackend(new MemoryStateBackend());
    //env.setStateBackend(new FsStateBackend("hdfs://zzy:9000/flink/checkpoints"));
    //rocksDB需要引入依赖flink-statebackend-rocksdb_2.11
    //env.setStateBackend(new RocksDBStateBackend("hdfs://zzy:9000/flink/checkpoints",true));
    env.setStateBackend(new FsStateBackend("hdfs://192.168.5.63:9000/flink/checkpoints"));

但是JobManager的web ui上checkpoint并未触发

![format_png 2][]

报错如下，应该是连接不到zzy 9001，识别不了zzy

![format_png 3][]

选择监听50.63上的9001端口，如果没有nc命令，用

yum install -y nc

安装下，用下面的命令启动flink程序，采用flink on yarn的方式

./bin/flink run -m yarn-cluster -yn 1 -yjm 1024 -ytm 1024 -c com.zzy.bigdata.flink.SocketWindowWordCountJavaCheckPoint zzy_flink_learn.jar --port 9001

![format_png 4][]

2019-03-06 16:00:24,680 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deployment took more than 60 seconds. Please check if the requested resources are available in the YARN cluster

如果一直出现Deployment xxx，此时可能是集群上没有资源了，  
这里杀掉application\_1551789318445\_0007和application\_1551789318445\_0008（这两台是测试机器，资源很紧张）

![format_png 5][]

然后再次重启程序

![format_png 6][]

注意yarn是不是successfully.的状态

![format_png 7][]

Yarn上启动了应用application\_1551789318445\_0009  
点击AM进去jobManager的web ui界面

![format_png 8][]

![format_png 9][]

![format_png 10][]

![format_png 11][]

Checkpoint的UI

![format_png 12][]

![format_png 13][]

可以看到每隔10s进行一次checkpoint

![format_png 14][]

![format_png 15][]

![format_png 16][]

Hdfs上查看checkpoint数据，看到保存了最近10次的checkpoint数据

![format_png 17][]

95d75e802ba1eceefeaf98636e907883跟job ID是对应的

![format_png 18][]

说明flink配置文件conf/flink-conf.yaml里的配置生效了

# flink可以保存多个checkpoint,添加如下配置，指定最多需要保存Checkpoint的个数 #

state.checkpoints.num-retained: 10

![format_png 19][]

**If nothing else is configured, the system will use the MemoryStateBackend.**

[https://www.jianshu.com/p/3cd2ab1dd311][https_www.jianshu.com_p_3cd2ab1dd311]

[https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state\_backends.html][https_ci.apache.org_projects_flink_flink-docs-release-1.8_ops_state_state_backends.html]

[format_png]: /images/20210920/5ad24c5a7c5c439d9e6f7f49251cfc37.png
[format_png 1]: /images/20210920/3b15d176090a4705834f0980d1e8801d.png
[format_png 2]: /images/20210920/f1d2969ba03d489b9af824d0dc7b6daa.png
[format_png 3]: /images/20210920/9ee32e209e3046dba5aabf8d3f250e26.png
[format_png 4]: /images/20210920/a62e5a33453b49058e89fd09d87663ea.png
[format_png 5]: /images/20210920/5ad928af612b4cb093955b67a5525b83.png
[format_png 6]: /images/20210920/6b2780532fe54fec8295411726138b12.png
[format_png 7]: /images/20210920/2bc48906f9b74f94bc3535f127605cf4.png
[format_png 8]: /images/20210920/2a112df51dc44f79b99499d4c7e51255.png
[format_png 9]: /images/20210920/d14c25142e7242a18ddb8535705976e0.png
[format_png 10]: /images/20210920/cbd67ca2d75049ba98b6348568220779.png
[format_png 11]: /images/20210920/96c1a8e13b374747a13021a0f569eb8d.png
[format_png 12]: /images/20210920/4c9f8fbb5c444b74b7cd0ec86e977710.png
[format_png 13]: /images/20210920/a6028e4144b2467babf0b298e5825d99.png
[format_png 14]: /images/20210920/984a4646c56f4fb1ae773ef9ab41354a.png
[format_png 15]: /images/20210920/8ffc689836834407b49fd4e3b50e969a.png
[format_png 16]: /images/20210920/70b0ac0af8634edf87debd7f495a584f.png
[format_png 17]: /images/20210920/6fabdd21fa844de783844034f7ce352f.png
[format_png 18]: /images/20210920/09e2fe91b9a1413088c933d8d7272691.png
[format_png 19]: /images/20210920/c27229534246492eb6b15be7e9cd0d68.png
[https_www.jianshu.com_p_3cd2ab1dd311]: https://www.jianshu.com/p/3cd2ab1dd311
[https_ci.apache.org_projects_flink_flink-docs-release-1.8_ops_state_state_backends.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html