- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

07_Flink之checkpoint和state

bigdata张凯翔发表于 2021/03/26 01:02:40 2021/03/26

【摘要】 1Flinkchecpoint容错机制【并行barrier面试必问】 image.png image.png barrier携带者快照的id，快照的id在快照存放在最前面 flink容错-checkpoint概念 image.png flink容错-checkpoint流程第二个问题：停止数据处理，做完快照，广播屏障。 1、快照数据存储，...

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

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

image.png

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

image.png

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

image.png

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

image.png

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

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;