当前位置: 首页 > news >正文

Flink源码之StreamTask启动流程

每个ExecutionVertex分配Slot后,JobMaster就会向Slot所在的TaskExecutor提交RPC请求执行Task,接口为TaskExecutorGateway::submitTask

CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, @RpcTimeout Time timeout); 

TaskDeploymentDescriptor 中包含当前Task的执行逻辑、Job信息、输入输出信息
在这里插入图片描述

submitTask 方法核心就是构造org.apache.flink.runtime.taskmanager.Task实例,该实例继承自Runnable接口,有个Thread成员变量,构造完成后就启动线程执行Task逻辑。

TaskExecutor::submitTask
Task.startTaskThread
Task.run
Task.doRun
Task::setupPartitionsAndGates //初始化Task的输入输出
RuntimeEnvironment::new //封装task执行上下文信息
Task::loadAndInstantiateInvokable //TaskInvokables实例化
StreamTask::newStreamTask::createRecordWriterDelegate //创建Writer,为每个StreamEdge创建一个WriterStreamTask::createStateBackend //创建StateBackend,一个task一个StateBackend实例StreamTask::createCheckpointStorageSubtaskCheckpointCoordinatorImpl::new 
Task::restoreAndInvoke
TaskInvokable::restore 
TaskInvokable::invoke //处理输入元素
TaskInvokable::cleanUp

Task的Invokable Class是在StreamGraph中添加Operator形成StreamNode时确定的,对不同的算子有不同的InvokableClass:

  • SourceStreamTask.class //LegacySource算子
  • SourceOperatorStreamTask //Source算子
  • OneInputStreamTask.class //输入是一个算子
  • TwoInputStreamTask:class //输入是两个算子
  • MultipleInputStreamTask.class //输入有多个算子

以上这些类都继承自org.apache.flink.streaming.runtime.tasks.StreamTask

在这里插入图片描述

在调用TaskInvokable::restore时会执行:

StreamTask::restore
StreamTask::restoreInternal //创建OperatorChain
RegularOperatorChain::new
OperatorChain::new
OperatorChain::createOutputCollector
OperatorChain::createOperatorChain
OperatorChain::createOperator
StreamOperatorFactoryUtil.createOperator  //创建Operator,在每个算子的StreamConfig中定义了每个Operator具体类型,比如StreamMap, StreamFlatMap
SimpleOperatorFactory::createStreamOperator //创建StreamOperator包装了用户函数,, StreamOperator包装了代码中用户函数,会调用用户函数中的open/close等生命周期函数AbstractUdfStreamOperator::setupAbstractStreamOperator::setup //设置用用自定义函数中的RuntimeContext成员变量StreamingRuntimeContext::new  //StreamTask::init //子类做初始化,创建InputGate、StreamTaskInput、DataOutput及InputProcessor
StreamTask::restoreGatesStreamTask::createStreamTaskStateInitializerStreamTaskStateInitializerImpl::new //OperatorChain::initializeStateAndOpenOperators //调用每个Operator的initializeState和Open方法AbstractStreamOperator::initializeState StreamTaskStateInitializerImpl::streamOperatorStateContext //此时会创建keyedStatedBackend和operatorStateBackendStreamOperatorStateHandler::new //初始化StreamOperator的stateHandler成员变量,用于状态管理StreamOperatorStateHandler::initializeOperatorStateStateInitializationContextImpl::newAbstractUdfStreamOperator::initializeState//调用用户定义函数中的initializeState方法,可获取Operator StateStreamingFunctionUtils::restoreFunctionStateStreamingRuntimeContext::setKeyedStateStoreStreamOperator::open //调用getRuntimeContext().getState可获取keySate
StreamTask::invoke
StreamTask::runMailboxLoop
MailboxProcessor::runMailboxLoop
StreamTask::processInput

整个过程在StreamTask.java的注释中有说明:

 * -- invoke()*       |*       +----> Create basic utils (config, etc) and load the chain of operators*       +----> operators.setup()*       +----> task specific init()*       +----> initialize-operator-states()*       +----> open-operators()*       +----> run()*       +----> finish-operators()*       +----> close-operators()*       +----> common cleanup*       +----> task specific cleanup()
  1. 首先创建OperatorChain,依次创建出每个StreamOperator
  2. 调用Operator的setup方法,初始化StreamingRuntimeContext
  3. 调用子类init方法初始化
  4. 调用initializeState初始化每个算子的状态,此时会为每个StreamOperator创建keyedStatedBackend和operatorStateBackend,然后会调用用户定义函数中的initializeState方法,用于创建Operator State
  5. 调用算子的open方法,便于用户在自定义函数open中进行初始化,比如初始化keyState
  6. 调用processInput处理流中数据

SourceStreamTask重载了StreamTask::processInput,该方法中直接起一个线程调用SourceFunction::run方法。

OneInputStreamTask则不同,它重载了StreamTask的init方法,在init方法中创建了StreamOneInputProcessor

OneInputStreamTask::init
OneInputStreamTask::createCheckpointedInputGate
OneInputStreamTask::createDataOutput //创建StreamTaskNetworkOutput
OneInputStreamTask::createTaskInput //创建StreamTaskNetworkInput
StreamOneInputProcessor::new

在StreamTask::processInput则是调用InputProcessor::processInput不断读取数据进行处理

StreamOneInputProcessor::processInput
StreamTaskNetworkInput::emitNext(StreamTaskNetworkOutput)
AbstractStreamTaskNetworkInput::emitNext //循环不断从buffer中读取StreamElement
处理
AbstractStreamTaskNetworkInput::processElementStreamTaskNetworkOutput::emitRecord //调用operator的setKeyContextElement和processElementOneInputStreamOperator::setKeyContextElementAbstractStreamOperator::setKeyContextElement1AbstractStreamOperator::setCurrentKey //StreamOperatorStateHandler::setCurrentKey //设置状态当前keyInput::processElement  //调用StreamOperator的processElement方法

以上Task从提交到起线程执行起来的整个过程,在初始化过程中为每个StreamOperator进行状态后端的初始化相当重要,后续处理流的过程中会使用这些状态后端存储管理状态。

相关文章:

Flink源码之StreamTask启动流程

每个ExecutionVertex分配Slot后&#xff0c;JobMaster就会向Slot所在的TaskExecutor提交RPC请求执行Task&#xff0c;接口为TaskExecutorGateway::submitTask CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, RpcTi…...

【BASH】回顾与知识点梳理(三十)

【BASH】回顾与知识点梳理 三十 三十. 进程的观察30.1 ps &#xff1a;将某个时间点的进程运作情况撷取下来仅观察自己的 bash 相关进程&#xff1a; ps -l观察系统所有进程&#xff1a; ps aux 30.2 top&#xff1a;动态观察进程的变化30.3 pstree 该系列目录 --> 【BASH】…...

亿赛通电子文档安全管理系统任意文件上传漏洞复现

0x01 产品简介 亿赛通电子文档安全管理系统&#xff08;简称&#xff1a;CDG&#xff09;是一款电子文档安全加密软件&#xff0c;该系统利用驱动层透明加密技术&#xff0c;通过对电子文档的加密保护&#xff0c;防止内部员工泄密和外部人员非法窃取企业核心重要数据资产&…...

java:数据库连接池

概念 举个例子来说吧&#xff0c;假设我们开了一家餐馆&#xff0c;客人来了&#xff0c;我们就请一个服务员&#xff0c;使用完后再把他开除了&#xff0c;下个客人再来了&#xff0c;我们再请一个&#xff0c;使用完再开除。 这是不是我们现在使用 JDBC 连接数据库的场景&a…...

可视化绘图技巧100篇基础篇(三)-条形图(一)

目录 前言 适用场景 图例 条形图分类 多系列条形图 单系列条形图...

如何使用Redis实现附近商家查询

导读 在日常生活中&#xff0c;我们经常能看见查询附近商家的功能。 常见的场景有&#xff0c;比如你在点外卖的时候&#xff0c;就可能需要按照距离查询附近几百米或者几公里的商家。 本文将介绍如何使用Redis实现按照距离查询附近商户的功能&#xff0c;并以SpringBoot项目…...

于vue3+vite+element pro + pnpm开源项目

河码桌面是一个基于vue3viteelement pro pnpm 创建的monorepo项目&#xff0c;项目采用的是类操作系统的web界面&#xff0c;操作起来简单又方便&#xff0c;符合用户习惯&#xff0c;又没有操作系统的复杂&#xff01; 有两个两个分支&#xff0c;一个是web版本&#xff0c;…...

18-组件化开发 根组件

组件化开发 & 根组件: 1. 组件化:一个页面可以拆分成一个个组件&#xff0c;每个组件有着自己独立的结构、样式、行为. 好处:便于维护&#xff0c;利于复用->提升开发效率 组件分类: 普通组件 , 根组件 2. 根组件:整个应用最上层的组件&#xff0c;包裹所有普通小组件…...

springboot集成ES

1.引入pom依赖2.application 配置3.JavaBean配置以及ES相关注解 3.1 Student实体类3.2 Teacher实体类3.3 Headmaster 实体类4. 启动类配置5.elasticsearchRestTemplate 新增 5.1 createIndex && putMapping 创建索引及映射 5.1.1 Controller层5.1.2 service层5.1.3 ser…...

Maven 生成编译时间和版本Java类

本文使用Maven插件来自动生成一个 Version.java 类&#xff0c;可以在Java代码中使用里面对应的常量&#xff0c;获取当前版本号和构建时间。 Maven编译后自动生成的 Version.java 文件内容如下所示&#xff1a; package com.shanhy.demo;public final class Version {public…...

关于uniapp微信小程序scroll-view组件使用show-scrollbar隐藏不了滚动条

这里关于使用 scroll-view组件 时候有滚动条 想要隐藏滚动条但是使用show-scrollbar没有效果 这时候又使用类名隐藏滚动条 使用id隐藏滚动条都不行 解决方法&#xff1a;在使用 scroll-view组件 的页面或者app 页面加上以下代码就可以了 ::-webkit-scrollbar {displa…...

CSS:filter滤镜 详解(用法 + 代码 + 例子 + 效果)

文章目录 filter 滤镜blur() 模糊度例子 渐变光晕 brightness() 元素亮度contrast() 对比度grayscale() 元素灰度hue-rorate() 色相opacity() 透明度invert() 反转颜色saturate() 饱和度 backdrop-filter 蒙版&#xff0c;滤镜例子 卷轴展开 filter 滤镜 动图为效果添加前后对…...

【Unity每日一记】Physics.Raycast 相关_Unity中的“X光射线”

&#x1f468;‍&#x1f4bb;个人主页&#xff1a;元宇宙-秩沅 &#x1f468;‍&#x1f4bb; hallo 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! &#x1f468;‍&#x1f4bb; 本文由 秩沅 原创 &#x1f468;‍&#x1f4bb; 收录于专栏&#xff1a;uni…...

软件报错msvcr90.dll丢失的解决方法,亲测可以修复

我曾经遇到过一个令人头疼的问题&#xff1a;msvcr90.dll丢失。这个问题导致了我的程序无法正常运行&#xff0c;让我感到非常苦恼。然而&#xff0c;在经过一番努力后&#xff0c;我终于成功地修复了这个问题&#xff0c;这让我感到非常欣慰和满足。 msvcr90.dll丢失的原因可能…...

第一百一十八回 如何获取蓝牙连接状态

文章目录 知识回顾实现方法示例代码我们在上一章回中介绍了如何连接蓝牙设备相关的内容,本章回中将介绍如何获取蓝牙连接状态.闲话休提,让我们一起Talk Flutter吧。 知识回顾 我们在上一章回中介绍如何连接蓝牙设备,但是如何知道蓝牙设备是否连接成功呢?这就需要我们去获取…...

C++11并发与多线程笔记(12) windows临界区、其他各种mutex互斥量

C11并发与多线程笔记&#xff08;12&#xff09; windows临界区、其他各种mutex互斥量 1、windows临界区2、自动析构技术3、递归独占互斥量 std::recursive_mutex4、带超时的互斥量 std::timed_mutex 和 std::recursive_timed_mutex4.1 std::timed_mutex&#xff1a;是带超时的…...

[MAUI]在.NET MAUI中实现可拖拽排序列表

文章目录 创建可拖放控件创建绑定服务类拖拽&#xff08;Drag&#xff09;拖拽悬停&#xff0c;经过&#xff08;DragOver&#xff09;释放&#xff08;Drop&#xff09; 创建页面元素最终效果项目地址 .NET MAUI 中提供了拖放(drag-drop)手势识别器&#xff0c;允许用户通过拖…...

Linux 内核内存管理 pfn_to_online_page宏

文章目录 一、Memory Hotplug1.1 简介1.2 热插拔事件通知机制 二、pfn_to_online_page2.1 pfn_to_online_page2.2 pfn_to_section_nr2.3 online_section_nr 参考资料 一、Memory Hotplug 1.1 简介 Linux 内存热插拔&#xff08;Memory Hotplug&#xff09;是指在运行时动态增…...

『C语言初阶』第八章 -结构体

前言 今天小羊又来给铁汁们分享关于C语言的结构体&#xff0c;在C语言中&#xff0c;结构体类型属于一种构造类型&#xff08;其他的构造类型还有&#xff1a;数组类型&#xff0c;联合类型&#xff09;&#xff0c;今天我们主要简单了解一下结构体。 一、结构体是什么&#x…...

chatgpt汇编,两个关于chatgpt的问题

1.用C写一个高级的可自主学习的AI语言模型 编写一个高级的可自主学习的AI语言模型是一个复杂且庞大的工程&#xff0c;需要深入的专业知识和大量的时间精力。下面是一个简化的示例&#xff0c;演示如何使用C实现一个基本的可自主学习的AI语言模型。 首先&#xff0c;你需要定…...

macOS效率工具:Dozer极简菜单栏管理方案

macOS效率工具&#xff1a;Dozer极简菜单栏管理方案 【免费下载链接】Dozer Hide menu bar icons on macOS 项目地址: https://gitcode.com/gh_mirrors/do/Dozer 在现代工作环境中&#xff0c;macOS用户常常面临菜单栏图标过多导致的视觉混乱问题。随着各类应用程序的安…...

AIGlasses_for_navigation 开发环境快速配置:Anaconda虚拟环境指南

AIGlasses_for_navigation 开发环境快速配置&#xff1a;Anaconda虚拟环境指南 你是不是也遇到过这种情况&#xff1a;好不容易在本地跑通了一个项目&#xff0c;换台电脑或者更新了几个库&#xff0c;结果就报了一堆莫名其妙的错误。或者&#xff0c;你想同时维护两个需要不同…...

RMBG-2.0隐私安全优势解析:纯本地推理杜绝图片上传与数据泄露风险

RMBG-2.0隐私安全优势解析&#xff1a;纯本地推理杜绝图片上传与数据泄露风险 在数字时代&#xff0c;图片处理已经成为日常工作和创作的重要组成部分&#xff0c;但随之而来的隐私安全问题也日益凸显。许多在线抠图工具需要用户上传图片到云端服务器进行处理&#xff0c;这不…...

别再死磕MIG了!ZYNQ PS端DDR3做帧缓存,用VDMA+HP接口实战指南

ZYNQ视频处理架构革命&#xff1a;VDMAHP接口实战全解析 从传统FPGA到ZYNQ的思维转换 在传统FPGA视频处理项目中&#xff0c;工程师们早已习惯使用MIG IP核管理DDR控制器&#xff0c;通过用户接口实现帧缓存功能。这种模式在纯FPGA环境中运行良好&#xff0c;但当转向ZYNQ平台…...

PDF文本高效提取:用pdftotext实现秒级文档内容解析

PDF文本高效提取&#xff1a;用pdftotext实现秒级文档内容解析 【免费下载链接】pdftotext Simple PDF text extraction 项目地址: https://gitcode.com/gh_mirrors/pd/pdftotext 破解PDF提取痛点&#xff1a;为什么你需要专业工具&#xff1f; 每天面对数十份PDF文档却…...

【SpringAI篇04】:从内存到MySQL,构建可重启的智能对话系统

1. 为什么需要从内存存储升级到数据库持久化 刚开始接触SpringAI开发时&#xff0c;很多开发者都会选择默认的内存存储方案。这种方案简单直接&#xff0c;不需要额外配置数据库&#xff0c;特别适合快速原型开发。但当你真正要把应用部署到生产环境时&#xff0c;就会发现内存…...

Qwen2-VL-2B-Instruct在Qt桌面应用中的集成:开发跨平台图像分析工具

Qwen2-VL-2B-Instruct在Qt桌面应用中的集成&#xff1a;开发跨平台图像分析工具 1. 引言 如果你是做桌面应用开发的&#xff0c;特别是用C和Qt的&#xff0c;最近可能也注意到了AI模型带来的新机会。很多开发者都在想&#xff0c;怎么把这些强大的AI能力&#xff0c;比如看图…...

重庆银行:万亿新贵的高光与隐忧

对于重庆银行而言&#xff0c;2026年3月24日是一个值得载入史册的日子。就在这一天&#xff0c;该行正式发布了2025年年度报告&#xff0c;其资产规模突破以往周期&#xff0c;使其成功跻身“万亿级城商行俱乐部”。其中&#xff0c;该行的营收与净利润时隔五年再次实现了“双十…...

突破设备边界:开源串流解决方案Sunshine革新跨设备游戏共享体验

突破设备边界&#xff1a;开源串流解决方案Sunshine革新跨设备游戏共享体验 【免费下载链接】Sunshine Sunshine: Sunshine是一个自托管的游戏流媒体服务器&#xff0c;支持通过Moonlight在各种设备上进行低延迟的游戏串流。 项目地址: https://gitcode.com/GitHub_Trending/…...

成长规划师 - OpenClaw助力个人发展

每周进步1%&#xff0c;一年后你会比现在优秀37倍你有没有过这样的感觉&#xff1a; 一周忙忙碌碌&#xff0c;周五回顾时却想不起做了什么重要的事&#xff1f;年初立下的flag&#xff0c;到了年底发现一个都没实现&#xff1f;羡慕别人技能满满&#xff0c;自己却不知道从哪里…...