07_Flink之checkpoint和state

1Flinkchecpoint容错机制【并行barrier面试必问】

image.png

barrier携带者快照的id，快照的id在快照存放在最前面

flink容错-checkpoint概念

image.png

flink容错-checkpoint流程
第二个问题：停止数据处理，做完快照，广播屏障。
1、快照数据存储，不往下发？只发barrier，应该里面包含着数据才对。
2、一个算子存储做，那么计算是存储之前做，还是存储之后做。？

image.png

第三个问题：最下面那个到底对吗

image.png

package cn.itzkx.helloflink.checkpoint

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.scala._

object Demo01_Checkpoint {

  def main(args: Array[String]): Unit = {
    //基于Flink流环境的Checkpoint
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置Checkpoint相关参数
    //1.先开启Checkpoint相关功能
    //设置每个5秒钟做一次快照
    env.enableCheckpointing(5000)
    //2.设置参数
    //设置语义为仅一次
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //设置快照制作的超时时间
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    //如果快照制作失败,是否让整体任务失败.默认值为true
    env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
    //3.设置存储路径
    env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink-checkpointing"))


    //编写正常业务代码
    val sourceStream: DataStream[String] = env.socketTextStream("node01", 9999)
    //数据转换 hello world java
    val mapData: DataStream[(String, Int)] = sourceStream.map(line => (line, 1))
    //数据分组
    val groupStream: KeyedStream[(String, Int), Tuple] = mapData.keyBy(0)
    //累加
    groupStream.sum(1).print()
    //执行程序
    env.execute()

  }
}

持久化存储

image.png

持久化存储

MemoryStateBackend

state数据保存在java堆内存中，执行checkpoint的时候，会把state的快照数据保存到jobmanager的内存中基于内存的state backend在生产环境下不建议使用。【因为怕丢失】

FsStateBackend

state数据保存在taskmanager的内存中，执行checkpoint的时候，会把state的快照数据保存到配置的文件系统中，可以使用hdfs等分布式文件系统。

RocksDBStateBackend

基于RocksDB + FS
RocksDB跟上面的都略有不同，它会在本地文件系统中维护状态，state会直接写入本地rocksdb中。同时RocksDB需要配置一个远端的filesystem。

代码:
val env = StreamExecutionEnvironment.getExecutionEnvironment
// start a checkpoint every 1000 ms
//开启Flink的Checkpoint
env.enableCheckpointing(5000)
// advanced options:
// 设置checkpoint的执行模式，最多执行一次或者至少执行一次
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// 设置checkpoint的超时时间
env.getCheckpointConfig.setCheckpointTimeout(60000)
// 如果在只做快照过程中出现错误，是否让整体任务失败：true是 false不是
env.getCheckpointConfig.setFailTasksOnCheckpointingErrors(false)
//设置同一时间有多少 个checkpoint可以同时执行
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
//设置checkpoint路径
env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink_checkpoint0000000"))

checkpoint与state之间的关系

image.png

checkpoint是一个动词，如果顺利的话，会产生一个名词。

image.png

也就是说state被包含于state

image.png

hello1 world1
hello2 world2
之前看到过一次，state知道hello和world一共出现过一次。

image.png

state：流式计算中持久化的状态
state存放在local state backend。

image.png

在flink中，state可以分为这两种，一种是keyed State，一种是

image.png

hello只会在一个state的并发上面，不会去其他上面。

并行，多个节点一起计算。
并发，一个节点，多个线程计算。

image.png

hello1永远只会到并发的task上面去。
不去keyby也能拿到它.

image.png

当前kafka记录到哪，可以记录当前kafka的offset。

image.png

存储了相关的operator state相关信息。
fromelements在源码里面是正儿八经的operator state

@PublicEvolving
public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedFunction {

   private static final long serialVersionUID = 1L;

   /** The (de)serializer to be used for the data elements. */
   private final TypeSerializer<T> serializer;

   /** The actual data elements, in serialized form. */
   private final byte[] elementsSerialized;

   /** The number of serialized elements. */
   private final int numElements;

   /** The number of elements emitted already. */
   private volatile int numElementsEmitted;

   /** The number of elements to skip initially. */
   private volatile int numElementsToSkip;

   /** Flag to make the source cancelable. */
   private volatile boolean isRunning = true;

   private transient ListState<Integer> checkpointedState;

image.png

triggercheckpoint，当task收到所有barrier之后，会进行快照，再快照之前，将自己的输出继续传递barrier，并将自己的状态异步写入到持久化存储中。

image.png

我已经把快照做好，并且把元数据告诉你。

image.png

持久化到目录下面

image.png