【Flink状态管理（二）各状态初始化入口】状态初始化流程详解与源码剖析

文章目录

    • 1. 状态初始化总流程梳理
    • 2.创建StreamOperatorStateContext
    • 3. StateInitializationContext的接口设计。
    • 4. 状态初始化举例：UDF状态初始化

在TaskManager中启动Task线程后，会调用StreamTask.invoke()方法触发当前Task中算子的执行，在invoke()方法中会调用restoreInternal()方法，这中间包括创建和初始化算子中的状态数据。
另外在invoke中，可以通过判断任务状态来判断是否需要初始化状态。

        // Allow invoking method 'invoke' without having to call 'restore' before it.if (!isRunning) {LOG.debug("Restoring during invoke will be called.");restoreInternal();}

StreamTask调用initializeStateAndOpenOperators()方法对当前Task中所有算子的状态数据进行初始化。

RegularOperatorChain.
public void initializeStateAndOpenOperators(StreamTaskStateInitializer streamTaskStateInitializer) throws Exception {  Iterator var2 = this.getAllOperators(true).iterator();  while(var2.hasNext()) {  StreamOperatorWrapper<?, ?> operatorWrapper = (StreamOperatorWrapper)var2.next();  StreamOperator<?> operator = operatorWrapper.getStreamOperator();  operator.initializeState(streamTaskStateInitializer);  operator.open();  }  }

 
找到了算子状态初始化的位置，我们继续了解状态是如何初始化的。

1. 状态初始化总流程梳理

AbstractStreamOperator.initializeState中描述了状态初始化的总体流程，如下代码以及注释：

# AbstractStreamOperator.initializeStatepublic final void initializeState(StreamTaskStateInitializer streamTaskStateManager)  throws Exception {  //1. 获取类型序列化器final TypeSerializer<?> keySerializer =  config.getStateKeySerializer(getUserCodeClassloader());  //2. get containingTaskfinal StreamTask<?, ?> containingTask = Preconditions.checkNotNull(getContainingTask());  final CloseableRegistry streamTaskCloseableRegistry =  Preconditions.checkNotNull(containingTask.getCancelables());  //3. create StreamOperatorStateContextfinal StreamOperatorStateContext context =  streamTaskStateManager.streamOperatorStateContext(  getOperatorID(),  getClass().getSimpleName(),  getProcessingTimeService(),  this,  keySerializer,  streamTaskCloseableRegistry,  metrics,  config.getManagedMemoryFractionOperatorUseCaseOfSlot(  ManagedMemoryUseCase.STATE_BACKEND,  runtimeContext.getTaskManagerRuntimeInfo().getConfiguration(),  runtimeContext.getUserCodeClassLoader()),  isUsingCustomRawKeyedState());  //4. create stateHandlerstateHandler =  new StreamOperatorStateHandler(  context, getExecutionConfig(), streamTaskCloseableRegistry);  timeServiceManager = context.internalTimerServiceManager();  //5. initialize OperatorStatestateHandler.initializeOperatorState(this);  //6. set KeyedStateStore in runtimeContextruntimeContext.setKeyedStateStore(stateHandler.getKeyedStateStore().orElse(null));  
}

在StreamOperator初始化状态数据的过程中，首先从StreamTask中获取创建状态需要的组件，例如托管状态的管理后端KeyedStateBackend、OperatorStateBackend以及原生状态管理的KeyedStateInputs和OperatorStateInputs组件。

状态数据操作过程中使用的管理组件最终都会封装成StateInitializationContext并传递给子类使用，例如在AbstractUdfStreamOperator中，就会使用StateInitializationContext中的信息初始化用户定义的UDF中的状态数据。

2.创建StreamOperatorStateContext

接下来看如何在Task实例初始化时创建这些组件，并将其存储在StreamOperatorStateContext中供算子使用，如下代码：

StreamTaskStateInitializerImpl
@Override  
public StreamOperatorStateContext streamOperatorStateContext(  @Nonnull OperatorID operatorID,  @Nonnull String operatorClassName,  @Nonnull ProcessingTimeService processingTimeService,  @Nonnull KeyContext keyContext,  @Nullable TypeSerializer<?> keySerializer,  @Nonnull CloseableRegistry streamTaskCloseableRegistry,  @Nonnull MetricGroup metricGroup,  double managedMemoryFraction,  boolean isUsingCustomRawKeyedState)  throws Exception {  //1. 获取task实例信息TaskInfo taskInfo = environment.getTaskInfo();  OperatorSubtaskDescriptionText operatorSubtaskDescription =  new OperatorSubtaskDescriptionText(  operatorID,  operatorClassName,  taskInfo.getIndexOfThisSubtask(),  taskInfo.getNumberOfParallelSubtasks());  final String operatorIdentifierText = operatorSubtaskDescription.toString();  final PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates =  taskStateManager.prioritizedOperatorState(operatorID);  CheckpointableKeyedStateBackend<?> keyedStatedBackend = null;  OperatorStateBackend operatorStateBackend = null;  CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs = null;  CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs = null;  InternalTimeServiceManager<?> timeServiceManager;  try {  // 创建keyed类型的状态后端// -------------- Keyed State Backend --------------  keyedStatedBackend =  keyedStatedBackend(  keySerializer,  operatorIdentifierText,  prioritizedOperatorSubtaskStates,  streamTaskCloseableRegistry,  metricGroup,  managedMemoryFraction);  //创建operator类型的状态后端// -------------- Operator State Backend --------------  operatorStateBackend =  operatorStateBackend(  operatorIdentifierText,  prioritizedOperatorSubtaskStates,  streamTaskCloseableRegistry);  //创建原生类型状态后端// -------------- Raw State Streams --------------  rawKeyedStateInputs =  rawKeyedStateInputs(  prioritizedOperatorSubtaskStates  .getPrioritizedRawKeyedState()  .iterator());  streamTaskCloseableRegistry.registerCloseable(rawKeyedStateInputs);  rawOperatorStateInputs =  rawOperatorStateInputs(  prioritizedOperatorSubtaskStates  .getPrioritizedRawOperatorState()  .iterator());  streamTaskCloseableRegistry.registerCloseable(rawOperatorStateInputs);  //创建Internal Timer Service Manager// -------------- Internal Timer Service Manager --------------  if (keyedStatedBackend != null) {  // if the operator indicates that it is using custom raw keyed state,  // then whatever was written in the raw keyed state snapshot was NOT written            // by the internal timer services (because there is only ever one user of raw keyed            // state);            // in this case, timers should not attempt to restore timers from the raw keyed            // state.            final Iterable<KeyGroupStatePartitionStreamProvider> restoredRawKeyedStateTimers =  (prioritizedOperatorSubtaskStates.isRestored()  && !isUsingCustomRawKeyedState)  ? rawKeyedStateInputs  : Collections.emptyList();  timeServiceManager =  timeServiceManagerProvider.create(  keyedStatedBackend,  environment.getUserCodeClassLoader().asClassLoader(),  keyContext,  processingTimeService,  restoredRawKeyedStateTimers);  } else {  timeServiceManager = null;  }  // -------------- Preparing return value --------------  return new StreamOperatorStateContextImpl(  prioritizedOperatorSubtaskStates.getRestoredCheckpointId(),  operatorStateBackend,  keyedStatedBackend,  timeServiceManager,  rawOperatorStateInputs,  rawKeyedStateInputs);  } catch (Exception ex) {  。。。。
}

流程梳理：

  1. 从environment中获取TaskInfo，并基于Task实例创建OperatorSubtaskDescriptionText。Operator中Task实例的描述信息包含OperatorID、OperatorClassName等，最终用于创建OperatorStateBackend的状态存储后端。
  2. 创建KeyedStateBackend，KeyedStateBackend是KeyedState的状态管理后端，提供创建和管理KeyedState的方法。
  3. 创建OperatorStateBackend，OperatorStateBackend是OperatorState的状态管理后端，提供获取和管理OperatorState的接口。
  4. 创建KeyGroupStatePartitionStreamProvider实例，提供创建和获取原生KeyedState的方法。
  5. 创建StatePartitionStreamProvider实例，提供创建和获取原生OperatorState的方法。
  6. 将所有创建出来的托管状态管理后端keyedStatedBackend和operatorStateBackend、原生状态存储后端rawKeyedStateInputs和rawOperatorStateInputs及timeServiceManager实例，全部封装在StreamOperatorStateContextImpl上下文对象中，并返回给AbstractStreamOperator使用。

 
小结
StreamTaskStateInitializer.streamOperatorStateContext()方法包含创建托管状态和原生状态管理后端的全过程。StreamOperator的实现类能够从StreamOperatorStateContext中获取这些状态管理组件，并使用它们创建指定类型的状态，最终状态数据会存储在状态管理后端指定的物理介质上，例如堆内存或RocksDB。

StateInitializationContext会被用于算子和UserDefinedFunction中，实现算子或函数中的状态数据操作。

 

3. StateInitializationContext的接口设计。

StateInitializationContext接口同时继承了ManagedInitializationContext接口和FunctionInitializationContext接口。StateInitializationContext接口的默认实现类为StateInitializationContextImpl。
在这里插入图片描述

  1. ManagedInitializationContext接口提供了托管状态使用的KeyedStateStore和OperatorStateStore获取方法，即KeyedStateBackend和OperatorStateBackend的封装类。算子进行初始化时，会通过KeyedStateStore和OperatorStateStore提供的方法创建和管理指定类型的托管状态。

  2. FunctionInitializationContext提供了用户自定义函数状态数据初始化需要的方法。它和ManagedInitializationContext保持一致，这主要是为了和算子使用的上下文进行区分，但两者的操作基本一致。

  3. StateInitializationContext提供了对托管状态数据的管理，并在内部继承和拓展了获取及管理原生状态数据的方法，如getRawOperatorStateInputs()、getRawKeyedStateInputs()等

  4. StateInitializationContextImpl具备操作管理状态和原生状态的能力。基于它可以获取不同类型的状态管理后端，并基于状态管理操作状态数据。

在这里插入图片描述

 

4. 状态初始化举例：UDF状态初始化

在AbstractStreamOperator中调用initializeState(StateInitializationContext context)抽象方法初始化Operator中的状态。这里以AbstractUdfStreamOperator为例说明具体算子、UDF是如何进行状态初始化的。

AbstractUdfStreamOperator.initializeState()方法实际上调用了StreamingFunctionUtils.restoreFunctionState()方法对User-Defined Function中的状态数据进行初始化和恢复，实际上就是将上文创建的StateInitializationContext上下文信息提供给Function接口使用。

public void initializeState(StateInitializationContext context) throws Exception {super.initializeState(context);StreamingFunctionUtils.restoreFunctionState(context, userFunction);
}

恢复函数内部的状态数据涉及Checkpoint的实现，我们会在之后介绍如何在StreamingFunctionUtils.restoreFunctionState()方法中恢复函数中的状态数据。

 
《Flink设计与实现：核心原理与源码解析》张利兵

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/254375.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Transformer实战-系列教程13：DETR 算法解读

Transformer实战-系列教程13：DETR 算法解读

&#x1f6a9;&#x1f6a9;&#x1f6a9;Transformer实战-系列教程总目录 有任何问题欢迎在下面留言 本篇文章的代码运行界面均在Pycharm中进行 本篇文章配套的代码资源已经上传 点我下载源码 1、物体检测 说到目标检测你能想到什么 faster-rcnn系列&#xff0c;开山之作&…
阅读更多...
Conda历史版本下载地址和python对应关系

Conda历史版本下载地址和python对应关系

一、前言 因为Conda安装版本问题&#xff0c;带来了很多问题&#xff0c;虽然不能直接确定二者之间的关系&#xff0c;但是安装指定版本的conda,确实是一个比较好的方法。特此记忆。 二、下载地址 下载最新版本&#xff1a;Free Download | Anaconda 下载历史版本&#xff…
阅读更多...
Springboot集成flowable工作流

Springboot集成flowable工作流

文章目录 概要包依赖问题小结 概要 Spirng boot 集成flowable <spring-boot.version>3.1.5</spring-boot.version>包依赖 <dependency><groupId>org.flowable</groupId><artifactId>flowable-spring-boot-starter</artifactId>&l…
阅读更多...
vuecli3 执行 npm run build 打包命令报错：TypeError: file.split is not a function

vuecli3 执行 npm run build 打包命令报错：TypeError: file.split is not a function

问题 今天有个项目在打包的时候遇到了一个问题&#xff0c;就是执行 npm run build 命令的时候报错了&#xff0c;如下&#xff1a; 解决 我排查了一下&#xff0c;模拟代码如下&#xff1a;在打包的时候用了 MinChunkSizePlugin const webpack require("webpack"…
阅读更多...
c++设计模式之装饰器模式

c++设计模式之装饰器模式

作用 为现有类增加功能 案例说明 class Car { public:virtual void show()0; };class Bmw:public Car { public:void show(){cout<<"宝马汽车>>"<<endl;} };class Audi:public Car { public:void show(){cout<<"奥迪汽车>>&q…
阅读更多...
5G NR 信道号计算

5G NR 信道号计算

一、5G NR的频段 增加带宽是增加容量和传输速率最直接的方法&#xff0c;目前5G最大带宽将会达到400MHz&#xff0c;考虑到目前频率占用情况&#xff0c;5G将不得不使用高频进行通信。 3GPP协议定义了从Sub6G(FR1)到毫米波(FR2)的5G目标频谱。 其中FR1是5G的核心频段&#xff0…
阅读更多...
戴上HUAWEI WATCH GT 4，解锁龙年新玩法

戴上HUAWEI WATCH GT 4，解锁龙年新玩法

春节将至&#xff0c;华为WATCH GT 4作为一款颜值和实力并存的手表&#xff0c;能为节日增添了不少趣味和便利。无论你是钟情于龙年表盘或定制属于自己的表盘&#xff0c;还是过年用来抢红包或远程操控手机拍全家福等等&#xff0c;它都能成为你的“玩伴”。接下来&#xff0c;…
阅读更多...
STM32单片机基本原理与应用（四）

STM32单片机基本原理与应用（四）

直流电机驱动控制原理 1、电机正反转控制 在STM32中&#xff0c;直流电机的正反转控制主要通过改变电机输入电源的极性来实现。当电机的电压极性发生变化时&#xff0c;电机的旋转方向也会相应改变。在硬件电路中&#xff0c;可以通过继电器或晶体管等电子开关来切换电机的电源…
阅读更多...
vscode连接ssh报错

vscode连接ssh报错

关于vscode更新版本至1.86后&#xff0c;导致无法连接服务器问题的记录 原因&#xff1a;vscode1.86更新了对glibc的要求&#xff0c;需要最低2.28版本&#xff0c;导致各种旧版本的linux发行版&#xff08;比如最常见的centos 7&#xff09;都无法用remote-ssh来连接了&#…
阅读更多...
springboot微信小程序uniapp学习计划与日程管理系统

springboot微信小程序uniapp学习计划与日程管理系统

基于springboot学习计划与日程管理系统&#xff0c;确定学习计划小程序的目标&#xff0c;明确用户需求&#xff0c;学习计划小程序的主要功能是帮助用户制定学习计划&#xff0c;并跟踪学习进度。页面设计主要包括主页、计划学习页、个人中心页等&#xff0c;然后用户可以利用…
阅读更多...
LiteFlow规则引擎框架

LiteFlow规则引擎框架

LiteFlow规则引擎框架 Hi&#xff0c;我是阿昌&#xff0c;今天介绍一个规则引擎框架&#xff0c;LiteFlow&#xff1b; 一、前言 那首先得知道什么是规则引擎&#xff1f;规则引擎是 一种用于自动化处理业务规则的软件组件。 在软件行业中&#xff0c;规则引擎通常用于解决…
阅读更多...
Python中HTTP隧道的基本原理与实现

Python中HTTP隧道的基本原理与实现

HTTP隧道是一种允许客户端和服务器之间通过中间代理进行通信的技术。这种隧道技术允许代理服务器转发客户端和服务器之间的所有HTTP请求和响应&#xff0c;而不需要对请求或响应内容进行任何处理或解析。Python提供了强大的网络编程能力&#xff0c;可以使用标准库中的socket和…
阅读更多...
深度学习入门笔记（八）可以不断思考的模型：RNN与LSTM

深度学习入门笔记（八）可以不断思考的模型：RNN与LSTM

8.1 循环神经网络RNN 之前学到的 CNN 和全连接&#xff0c;模型的输入数据之间是没有关联的&#xff0c;比如图像分类&#xff0c;每次输入的图片与图片之间就没有任何关系&#xff0c;上一张图片的内容不会影响到下一张图片的结果。但在自然语言处理领域&#xff0c;这就成了…
阅读更多...
RabbitMQ高可用架构涉及常用功能整理

RabbitMQ高可用架构涉及常用功能整理

RabbitMQ高可用架构涉及常用功能整理 1. rabbitmq的集群模式2. 镜像模式高可用系统架构和相关组件3. rabbitmq的核心参数3.1 镜像策略3.2 新镜像同步策略3.3 从节点晋升策略3.4 主队列选择策略 4. rabbitmq常用命令4.1 常用基础命令4.1.1 服务管理4.1.2 用户管理4.1.3 角色管理…
阅读更多...
MacOS上怎么把格式化成APFS的U盘或者硬盘格式化回ExFAT？

MacOS上怎么把格式化成APFS的U盘或者硬盘格式化回ExFAT？

一、问题 MacOS在更新MacOS Monterey后或者更高系统后发现&#xff0c;格式U盘或者硬盘只有4个APFS选项&#xff0c;那么我们该如何将APFS格式成ExFAT&#xff1f; 二、解答 将APFS的U盘或者硬盘拓展成MacOS的拓展格式即可&#xff0c;操作步骤如下 1、电脑接入U盘或者硬盘 2…
阅读更多...
项目02《游戏-11-开发》Unity3D

项目02《游戏-11-开发》Unity3D

基于 项目02《游戏-10-开发》Unity3D &#xff0c; 任务&#xff1a;飞行坐骑 首先创建脚本&#xff0c; 绑定脚本&#xff0c; using UnityEngine; public class Dragon : MonoBehaviour{ [SerializeField] private float speed 10f; public Transfo…
阅读更多...
RCE（命令执行）知识点总结最详细

RCE（命令执行）知识点总结最详细

description: 这里是CTF做题时常见的会遇见的RCE的漏洞知识点总结。 如果你觉得写得好并且想看更多web知识的话可以去gitbook.22kaka.fun去看&#xff0c;上面是我写的一本关于web学习的一个gitbook&#xff0c;当然如果你能去我的github为我的这个项目点亮星星我会感激不尽htt…
阅读更多...
uniapp的配置和使用

uniapp的配置和使用

①安装环境和编辑器 注册小程序账号 微信开发者工具下载 uniapp 官网 HbuilderX 下载 首先先下载Hbuilder和微信开发者工具 &#xff08;都是傻瓜式安装&#xff09;&#xff0c;然后注册小程序账号&#xff1a; 拿到appid&#xff1a; ②简单通过demo使用微信开发者工具和…
阅读更多...
Flask基础学习

Flask基础学习

1.debug、host、port 模式修改 1) debug模式 默认debug模式是off&#xff0c;在修改代码调试过程中需要暂停重启使用&#xff0c;这时可修改on模式解决。 同时在debug模式开启下可看到出错信息。 下面有关于Pycharm社区版和专业版修改debug模式的区别 专业版 社区版&#…
阅读更多...
目标检测 | 卷积神经网络（CNN）详细介绍及其原理详解

目标检测 | 卷积神经网络（CNN）详细介绍及其原理详解

前言&#xff1a;Hello大家好&#xff0c;我是小哥谈。卷积神经网络&#xff08;Convolutional Neural Network&#xff0c;CNN&#xff09;是一种深度学习模型&#xff0c;主要用于图像识别和计算机视觉任务。它的设计灵感来自于生物学中视觉皮层的工作原理。CNN的核心思想是通…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023