文章目录
- 一. 状态后端概述
- 二. StateBackend的整体设计
-
- 1. 核心功能
- 2. StateBackend的UML
- 3. 小结
- 三. StateBackend的加载与初始化
-
- 1. StateBackend创建概述
- 2. StateBackend创建过程
一. 状态后端概述
StateBackend作为状态存储后端,提供了创建和获取KeyedStateBackend及OperatorStateBackend的方法,并通过CheckpointStorage实现了对状态数据的持久化存储。
Flink支持MemoryStateBackend、FsStateBackend和RocksDBStateBackend三种类型的状态存储后端,三者的主要区别在于创建的KeyedStateBackend及CheckpointStorage不同。例如,MemoryStateBackend和FileStateBackend创建的是HeapKeyedStateBackend,RocksDBStateBackend创建的是RocksDBKeyedStateBackend。
本文关注StateBackend的设计与实现。
二. StateBackend的整体设计
1. 核心功能
在StateBackend接口中提供了如下核心功能。
- resolveCheckpoint()方法用于获取Checkpoint的Location信息,Location信息包含Checkpoint元数据信息;
- createCheckpointStorage()方法为Job创建CheckpointStorage对象,CheckpointStorage提供写入Checkpoint数据和元数据信息的能力;
- createKeyedStateBackend()方法用于创建KeyedStateBackend,KeyedStateBackend提供创建和管理KeyedState的能力;
- createOperatorStateBackend()方法主要用于创建OperatorStateBackend,通过OperatorStateBackend可以创建和管理OperatorState状服务器托管网态数据。
2. StateBackend的UML
StateBackend主要有AbstractStateBackend基本实现类,该类中没有提供实质性的方法,主要为了向前(ing)兼容。
AbstractFileStateBackend有MemoryStateBackend和FsStateBackend两种实现类,其中MemoryStateBackend主要通过JobManager堆内存存储Checkpoint数据,FsStateBackend通过FsCheckpointStorage将Checkpoint数据存储在指定文件系统中。
RockdsDBStateBackend也实现了StateBackend的基本功能,
- 内存状态:和其他状态管理后端不同的是,它创建的KeyedStateBackend是基于RocksDB实现的RocksDBKeyedStateBackend。KeyedState数据都会存储在RocksDB内存中。
- 持久化:对于CheckpointStorage的创建,RocksDBStateBackend依赖于FsStateBackend,即基于文件系统对Checkpoint中的状态数据进行持久化。
3. 小结
StateBackend提供了创建CheckpointStorage、KeyedStateBackend及OperatorStateBackend的功能。
基于MemoryStateBackend可以实现非常高效的状态数据获取和存储,但由于JobManager内存数量有限,对比较大的状态数据无法提供更好的支持。对于RocksDBStateBackend而言,可以基于RocksDB提供的LSM-Tree(Log StructuredMerge-Tree)内存数据结构,实现更加高效的堆外内存访问,支持大数据量的状态数据存储,这对生产环境来讲是一个更优的选择。
三. StateBackend的加载与初始化
1. StateBackend创建概述
StateBackend主要通过StateBackendFactory接口创建。StateBackendFactory主要有MemoryStateBackendFactory、FsStateBackendFactory和RocksDBStateBackendFactory三种实现,最终通过StateBackendFactory的不同实现类创建相应的StateBackend。
StateBackendFactory主要通过StateBackendLoader进行加载和创建。StateBackendLoader会根据state.backend的名称使用Java SPI技术加载相应类型的StateBackendFactory,最终创建StateBackend。
2. StateBackend创建过程
StateBackend会在两个过程中创建:
- 首先,在JobMaster根据JobGraph对象创建ExecutionGraph的过程中会创建StateBackend,用于CheckpointCoordinator组件管理状态和Checkpoint操作;
- 其次,在每个Task实例初始化的过程中会创建StateBackend,用于管理当前Task中的状态和Checkpoint数据。
接下来我们分步骤看StateBackend的创建过程。
1)在StreamTask中初始化StateBackend
前面我们已经知道,当StreamTask在TaskManager的Task线程中启动时,会调用invoke()抽象方法运行StreamTask中的算子。此时在beforeInvoke()方法中就会调用StreamTask.createStateBackend()方法创建当前Task中使用的StateBackend。
在StreamTask.createStateBackend()方法中可以看出,
//
private StateBackend createStateBackend() throws Exception {
//1. 从UserCodeClassLoader获取StateBackend
final StateBackend fromAppl服务器托管网ication =
configuration.getStateBackend(getUserCodeClassLoader());
//2. 通过应用配置还是通过集群默认配置创建StateBackend
return StateBackendLoader.fromApplicationOrConfigOrDefault(
fromApplication,
getEnvironment().getTaskManagerInfo().getConfiguration(),
getUserCodeClassLoader(),
LOG);
//用户在代码中调用StreamExecutionEnvironment.enableCheckpointing()方法时,
//系统默认配置主要是通过flink-conf.yaml启用StateBackend配置项。
}
2)StateBackendLoader加载配置的StateBackend
public static StateBackend fromApplicationOrConfigOrDefault(
@Nullable StateBackend fromApplication,
Configuration config,
ClassLoader classLoader,
@Nullable Logger logger)
throws IllegalConfigurationException, DynamicCodeLoadingException, IOException {
checkNotNull(config, "config");
checkNotNull(classLoader, "classLoader");
final StateBackend backend;
// 1) 如果应用配置的StateBackend不为空,则最高优先级是应用中定义的
//StateBackend实现类。
if (fromApplication != null) {
if (logger != null) {
logger.info("Using application-defined state backend: {}", fromApplication);
}
// 向fromApplication中追加额外的参数配置
if (fromApplication instanceof ConfigurableStateBackend) {
if (logger != null) {
logger.info("Configuring application-defined state backend with
job/cluster config");
}
// 直接从UserClassLoader中反序列化出StateBackend
backend = ((ConfigurableStateBackend) fromApplication)
.configure(config, classLoader);
}
else {
backend = fromApplication;
}
}
else {
//检查是否开启StateBackend默认配置
final StateBackend fromConfig =
loadStateBackendFromConfig(config, classLoader, logger);
if (fromConfig != null) {
backend = fromConfig;
} else {
//2. 如果配置为空则,创建默认MemoryStateBackend
backend = new MemoryStateBackendFactory().createFromConfig(config,
classLoader);
if (logger != null) {
logger.info("No state backend has been configured, using default
(Memory / JobManager) {}", backend);
}
}
}
return backend;
}
3)通过StateBackendFactory创建StateBackend
这里举例说明MemoryStateBackend的创建过程。从方法中,调用了MemoryStateBackend()构造器创建基于堆内存的StateBackend,并调用configure()方法对StateBackend进行参数配置。
public MemoryStateBackend createFromConfig(Configuration config, ClassLoader
classLoader) {
return new MemoryStateBackend().configure(config, classLoader);
}
《Flink设计与实现:核心原理与源码解析》–张利兵
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net