【flink状态管理(四)】MemoryStateBackend的实现
文章目录
- 1.基于MemoryStateBackend创建KeyedStateBackend
- 1.1. 状态初始化
- 1.2. 创建状态
- 2. 基于MemoryStateBackend创建OperatorStateBackend
- 3.基于MemoryStateBackend创建CheckpointStorage
在Flink中,默认的StateBackend实现为MemoryStateBackend,本文以MemoryStateBackend为例说明StateBackend的设计与实现。
本文介绍MemoryStateBackend中如下三个主要组件的创建过程:
- HeapKeyedStateBackend
- OperatorStateBackend
- MemoryBackendCheckpointStorage
FsStateBackend和RocksDBStateBackend这两种状态后端存储的实现,功能和MemoryStateBackend类似,区别在于内部创建的KeyedStateBackend和CheckpointStorage。
1.基于MemoryStateBackend创建KeyedStateBackend
1.1. 状态初始化
AbstractStreamOperator.keyedStatedBackend()方法定义了创建和初始化KeyedStatedBackend的逻辑,具体如下。
protected <K> AbstractKeyedStateBackend<K> keyedStateBackend(TypeSerializer<K> keySerializer,String operatorIdentifierText,PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates,CloseableRegistry backendCloseableRegistry,MetricGroup metricGroup) throws Exception {if (keySerializer == null) {return null;}String logDescription = "keyed state backend for " + operatorIdentifierText;//1. TaskInfo taskInfo = environment.getTaskInfo();final KeyGroupRange keyGroupRange = KeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(taskInfo.getMaxNumberOfParallelSubtasks(),taskInfo.getNumberOfParallelSubtasks(),taskInfo.getIndexOfThisSubtask());// 确保恢复状态过程中构建的数据流被关闭CloseableRegistry cancelStreamRegistryForRestore = new CloseableRegistry();backendCloseableRegistry.registerCloseable(cancelStreamRegistryForRestore);// 创建BackendRestorerProcedureBackendRestorerProcedure<AbstractKeyedStateBackend<K>, KeyedStateHandle> backendRestorer =new BackendRestorerProcedure<>((stateHandles) -> stateBackend.createKeyedStateBackend(environment,environment.getJobID(),operatorIdentifierText,keySerializer,taskInfo.getMaxNumberOfParallelSubtasks(),keyGroupRange,environment.getTaskKvStateRegistry(),TtlTimeProvider.DEFAULT,metricGroup,stateHandles,cancelStreamRegistryForRestore),backendCloseableRegistry,logDescription);try {return backendRestorer.createAndRestore(prioritizedOperatorSubtaskStates.getPrioritizedManagedKeyedState());} finally {if (backendCloseableRegistry.unregisterCloseable(cancelStreamRegistryForRestore)) {IOUtils.closeQuietly(cancelStreamRegistryForRestore);}}
}
- 获取当前Task的TaskInfo,并基于TaskInfo的参数创建KeyGroupRange,表示当前Task实例中存储的Key分组区间。
- 创建CloseableRegistry并注册到backendCloseableRegistry中,用于确保在任务取消的情况下关闭在恢复状态过程中构造的数据流。
- 创建BackendRestorerProcedure,提供了stateBackend.createKeyedStateBackend()方法,也包含恢复历史状态数据的方法。
- 创建KeyedStateBackend,同时对状态数据进行恢复。prioritizedOperatorSubtaskStates是从TaskStateManager中根据OperatorID获取的算子历史状态,通过prioritizedOperatorSubtaskStates获取当前算子的PrioritizedManagedKeyedState,并基于这些状态数据恢复算子的状态。
1.2. 创建状态
接下来我们看MemoryStateBackend.createKeyedStateBackend()方法的具体实现。
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(Environment env,JobID jobID,String operatorIdentifier,TypeSerializer<K> keySerializer,int numberOfKeyGroups,KeyGroupRange keyGroupRange,TaskKvStateRegistry kvStateRegistry,TtlTimeProvider ttlTimeProvider,MetricGroup metricGroup,@Nonnull Collection<KeyedStateHandle> stateHandles,CloseableRegistry cancelStreamRegistry) throws BackendBuildingException {// 获取TaskStateManager实例TaskStateManager taskStateManager = env.getTaskStateManager();// 创建HeapPriorityQueueSetFactory实例HeapPriorityQueueSetFactory priorityQueueSetFactory =new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);// 创建HeapKeyedStateBackendBuilder实例HeapKeyedStateBackendreturn new HeapKeyedStateBackendBuilder<>(kvStateRegistry,keySerializer,env.getUserClassLoader(),numberOfKeyGroups,keyGroupRange,env.getExecutionConfig(),ttlTimeProvider,stateHandles,AbstractStateBackend.getCompressionDecorator(env.getExecutionConfig()),taskStateManager.createLocalRecoveryConfig(),priorityQueueSetFactory,isUsingAsynchronousSnapshots(),cancelStreamRegistry).build();
}
- 从environment参数中获取TaskStateManager实例
- 创建HeapPriorityQueueSetFactory实例,用于生成HeapPriorityQueueSet优先级队列,存储TimerHeapInternalTimer等数据。
- 调用HeapKeyedStateBackendBuilder.build()方法创建HeapKeyedStateBackend。
2. 基于MemoryStateBackend创建OperatorStateBackend
和创建KeyedStateBackend的过程相似,AbstractStreamOperator.operatorStateBackend()方法实现了创建OperatorStateBackend的方法。
protected OperatorStateBackend operatorStateBackend(String operatorIdentifierText,PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates,CloseableRegistry backendCloseableRegistry) throws Exception {String logDescription = "operator state backend for " + operatorIdentifierText;CloseableRegistry cancelStreamRegistryForRestore = new CloseableRegistry();backendCloseableRegistry.registerCloseable(cancelStreamRegistryForRestore);BackendRestorerProcedure<OperatorStateBackend, OperatorStateHandle> backendRestorer =new BackendRestorerProcedure<>((stateHandles) -> stateBackend.createOperatorStateBackend(environment,operatorIdentifierText,stateHandles,cancelStreamRegistryForRestore),backendCloseableRegistry,logDescription);try {return backendRestorer.createAndRestore(prioritizedOperatorSubtaskStates.getPrioritizedManagedOperatorState());} finally {if (backendCloseableRegistry.unregisterCloseable(cancelStreamRegistryForRestore)) {IOUtils.closeQuietly(cancelStreamRegistryForRestore);}}
}
- 创建CloseableRegisty,确保在任务取消的情况下能够关闭在恢复状态时构造的数据流。
- 创建BackendRestorerProcedure,封装了stateBackend.createOperatorStateBackend()方法,并包含恢复历史状态数据的操作。
- 创建OperatorStateBackend,并恢复状态数据。
其中prioritizedOperatorSubtaskStates是从TaskStateManager中根据OperatorID获取的算子专有历史状态,可以通过prioritizedOperatorSubtaskStates获取当前算子中的PrioritizedManagedOperatorState,并基于这些状态数据恢复OperatorStateBackend中算子的状态。
3.基于MemoryStateBackend创建CheckpointStorage
在createCheckpointStorage()方法中,直接创建MemoryBackendCheckpointStorage实例并返回,没有涉及太多的流程
public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {return new MemoryBackendCheckpointStorage(jobId, getCheckpointPath(), getSavepointPath(), maxStateSize);
}
参考:《Flink设计与实现:核心原理与源码解析》–张利兵
相关文章:
【flink状态管理(四)】MemoryStateBackend的实现
文章目录 1.基于MemoryStateBackend创建KeyedStateBackend1.1. 状态初始化1.2. 创建状态 2. 基于MemoryStateBackend创建OperatorStateBackend3.基于MemoryStateBackend创建CheckpointStorage 在Flink中,默认的StateBackend实现为MemoryStateBackend,本文…...
前端架构: 脚手架在前端研发流程中的意义
关于脚手架 脚手架又被成为 CLI (command-line interface)基于文本界面,通过中断输入命令执行常见的脚手架:npm, webpack-cli, vue-cli拿 npm 这个脚手架来说 在终端当中输入 npm 命令, 系统就会通过文本方式返回 npm 的使用方法它这种通过命令行执行的…...
Qt网络编程-QTcpServer的封装
简单封装Tcp服务器类,将QTcpServer移入线程 头文件: #ifndef TCPSERVER_H #define TCPSERVER_H#include <QObject>class QTcpSocket; class QTcpServer; class QThread; class TcpServer : public QObject {Q_OBJECT public:explicit TcpServer(…...
【MySQL】_JDBC编程
目录 1. JDBC原理 2. 导入JDBC驱动包 3. 编写JDBC代码实现Insert 3.1 创建并初始化一个数据源 3.2 和数据库服务器建立连接 3.3 构造SQL语句 3.4 执行SQL语句 3.5 释放必要的资源 4. JDBC代码的优化 4.1 从控制台输入 4.2 避免SQL注入的SQL语句 5. 编写JDBC代码实现…...
微信小程序编译出现 project.config.json 文件内容错误
问题描述: 更新微信开发工具后,使用微信开发工具编译时出现project.config.json 文件内容错误。 原因:当前使用的微信开发工具非稳定版本。 解决方法: 在 manifest.json中加入以下代码: "mp-weixin" : …...
一周学会Django5 Python Web开发-Django5创建项目(用命令方式)
锋哥原创的Python Web开发 Django5视频教程: 2024版 Django5 Python web开发 视频教程(无废话版) 玩命更新中~_哔哩哔哩_bilibili2024版 Django5 Python web开发 视频教程(无废话版) 玩命更新中~共计11条视频,包括:2024版 Django5 Python we…...
DockerUI如何部署结合内网穿透实现公网环境管理本地docker容器
文章目录 前言1. 安装部署DockerUI2. 安装cpolar内网穿透3. 配置DockerUI公网访问地址4. 公网远程访问DockerUI5. 固定DockerUI公网地址 前言 DockerUI是一个docker容器镜像的可视化图形化管理工具。DockerUI可以用来轻松构建、管理和维护docker环境。它是完全开源且免费的。基…...
UML之在Markdown中使用Mermaid绘制类图
1.UML概述 UML(Unified modeling language UML)统一建模语言,是一种用于软件系统分析和设计的语言工具,它用于帮助软件开发人员进行思考和记录思路。 类图是描述类与类之间的关系的,是UML图中最核心的。类图的是用于…...
Spring Boot + 七牛OSS: 简化云存储集成
引言 Spring Boot 是一个非常流行的、快速搭建应用的框架,它无需大量的配置即可运行起来,而七牛云OSS提供了稳定高效的云端对象存储服务。利用两者的优势,可以为应用提供强大的文件存储功能。 为什么选择七牛云OSS? 七牛云OSS提供了高速的…...
C++:二叉搜索树模拟实现(KV模型)
C:二叉搜索树模拟实现(KV模型) 前言模拟实现KV模型1. 节点封装2、前置工作(默认构造、拷贝构造、赋值重载、析构函数等)2. 数据插入(递归和非递归版本)3、数据删除(递归和非递归版本…...
npm淘宝镜像源换新地址
新的淘宝npm镜像源地址:https://registry.npmmirror.com 切换新的镜像源 npm config set registry https://registry.npmmirror.com然后再执行以下操作查看是否成功 npm config list如果没安装过淘宝镜像源的,则直接安装 npm install -g cnpm --regi…...
十大排序算法之线性时间非比较类排序
线性时间非比较类排序 线性时间的算法执行效率也较高,从时间占用上看,线性时间非比较类排序要优于非线性时间排序,但其空间复杂度较非线性时间排序要大一些。因为线性时间非比较类排序算法会额外申请一定的空间进行分配排序,这也…...
容器基础:Docker 镜像如何保证部署的一致性?
Docker 镜像如何通过固化基础环境、固化依赖性和固化软件启动流程保证部署的一致性 Docker 镜像通过以下三个方面保证部署的一致性: 1. 固化基础环境: 镜像包含构建应用程序所需的所有环境依赖项,例如操作系统、库和工具。构建镜像时,所有…...
爪哇部落算法组2024新生赛热身赛题解
第一题(签到): 1、题意: 2、题解: 我们观察到happynewyear的长度是12个字符,我们直接从前往后遍历0到n - 12的位置(这里索引从0开始),使用C的substr()函数找到以i开头的长度为12的字…...
1123. 铲雪车(欧拉回路)
活动 - AcWing 随着白天越来越短夜晚越来越长,我们不得不考虑铲雪问题了。 整个城市所有的道路都是双向车道,道路的两个方向均需要铲雪。因为城市预算的削减,整个城市只有 1 辆铲雪车。 铲雪车只能把它开过的地方(车道)的雪铲干…...
网络协议与攻击模拟_15FTP协议
了解FTP协议 在Windows操作系统上使用serv-U软件搭建FTP服务 分析FTP流量 一、FTP协议 1、FTP概念 FTP(文件传输协议)由两部分组成:客户端/服务端(C/S架构) 应用场景:企业内部存放公司文件、开发网站时利…...
「效果图渲染」效果图与3D影视动画渲染平台
效果图渲染和3D影视动画渲染都是视觉图像渲染的领域应用。效果图渲染主要服务于建筑、室内设计和产品设计等行业,这些领域通常对视觉呈现的精度和细节有较高要求。与之相比,3D影视动画渲染则普遍应用于电影、电视、视频游戏和广告等媒体领域,…...
Blender_查看版本
Blender_查看版本 烦人的烦恼,没找见哪儿可以查看版本? 算是个隐蔽的角落!...
node.js 读目录.txt文件,用 xml2js 转换为json数据,生成jstree所需的文件
请参阅:java : pdfbox 读取 PDF文件内书签 请注意:书的目录.txt 编码:UTF-8,推荐用 Notepad 转换编码。 npm install elementtree ; npm install xml2js ; node.js 用 elementtree读目录.txt文件,用 xml2js 转换为…...
【Docker】02 镜像管理
文章目录 一、Images镜像二、管理操作2.1 搜索镜像2.1.1 命令行搜索2.1.2 页面搜索2.1.3 搜索条件 2.2 下载镜像2.3 查看本地镜像2.3.1 docker images2.3.2 --help2.3.3 repository name2.3.4 --filter2.3.5 -q2.3.6 --format 2.4 给镜像打标签2.5 推送镜像2.6 删除镜像2.7 导出…...
后进先出(LIFO)详解
LIFO 是 Last In, First Out 的缩写,中文译为后进先出。这是一种数据结构的工作原则,类似于一摞盘子或一叠书本: 最后放进去的元素最先出来 -想象往筒状容器里放盘子: (1)你放进的最后一个盘子(…...
Linux应用开发之网络套接字编程(实例篇)
服务端与客户端单连接 服务端代码 #include <sys/socket.h> #include <sys/types.h> #include <netinet/in.h> #include <stdio.h> #include <stdlib.h> #include <string.h> #include <arpa/inet.h> #include <pthread.h> …...
国防科技大学计算机基础课程笔记02信息编码
1.机内码和国标码 国标码就是我们非常熟悉的这个GB2312,但是因为都是16进制,因此这个了16进制的数据既可以翻译成为这个机器码,也可以翻译成为这个国标码,所以这个时候很容易会出现这个歧义的情况; 因此,我们的这个国…...
Java 8 Stream API 入门到实践详解
一、告别 for 循环! 传统痛点: Java 8 之前,集合操作离不开冗长的 for 循环和匿名类。例如,过滤列表中的偶数: List<Integer> list Arrays.asList(1, 2, 3, 4, 5); List<Integer> evens new ArrayList…...
postgresql|数据库|只读用户的创建和删除(备忘)
CREATE USER read_only WITH PASSWORD 密码 -- 连接到xxx数据库 \c xxx -- 授予对xxx数据库的只读权限 GRANT CONNECT ON DATABASE xxx TO read_only; GRANT USAGE ON SCHEMA public TO read_only; GRANT SELECT ON ALL TABLES IN SCHEMA public TO read_only; GRANT EXECUTE O…...
DBAPI如何优雅的获取单条数据
API如何优雅的获取单条数据 案例一 对于查询类API,查询的是单条数据,比如根据主键ID查询用户信息,sql如下: select id, name, age from user where id #{id}API默认返回的数据格式是多条的,如下: {&qu…...
UR 协作机器人「三剑客」:精密轻量担当(UR7e)、全能协作主力(UR12e)、重型任务专家(UR15)
UR协作机器人正以其卓越性能在现代制造业自动化中扮演重要角色。UR7e、UR12e和UR15通过创新技术和精准设计满足了不同行业的多样化需求。其中,UR15以其速度、精度及人工智能准备能力成为自动化领域的重要突破。UR7e和UR12e则在负载规格和市场定位上不断优化…...
华为OD机考-机房布局
import java.util.*;public class DemoTest5 {public static void main(String[] args) {Scanner in new Scanner(System.in);// 注意 hasNext 和 hasNextLine 的区别while (in.hasNextLine()) { // 注意 while 处理多个 caseSystem.out.println(solve(in.nextLine()));}}priv…...
热门Chrome扩展程序存在明文传输风险,用户隐私安全受威胁
赛门铁克威胁猎手团队最新报告披露,数款拥有数百万活跃用户的Chrome扩展程序正在通过未加密的HTTP连接静默泄露用户敏感数据,严重威胁用户隐私安全。 知名扩展程序存在明文传输风险 尽管宣称提供安全浏览、数据分析或便捷界面等功能,但SEMR…...
【实施指南】Android客户端HTTPS双向认证实施指南
🔐 一、所需准备材料 证书文件(6类核心文件) 类型 格式 作用 Android端要求 CA根证书 .crt/.pem 验证服务器/客户端证书合法性 需预置到Android信任库 服务器证书 .crt 服务器身份证明 客户端需持有以验证服务器 客户端证书 .crt 客户端身份…...
