Flink Job 执行流程

Flink On Yarn 模式

在这里插入图片描述

​

基于Yarn层面的架构类似 Spark on Yarn模式，都是由Client提交App到RM上面去运行，然后 RM分配第一个container去运行AM，然后由AM去负责资源的监督和管理。需要说明的是，Flink的Yarn模式更加类似Spark on Yarn的cluster模式，在cluster模式中，dirver将作为AM中的一个线程去运行。Flink on Yarn模式也是会将JobManager启动在container里面，去做个driver类似的任务调度和分配，Yarn AM与Flink JobManager在同一个Container中，这样AM可以知道Flink JobManager的地址，从而AM可以申请Container去启动Flink TaskManager。待Flink成功运行在Yarn集群上，Flink Yarn Client就可以提交Flink Job到Flink JobManager，并进行后续的映射、调度和计算处理。

Fink on Yarn 的缺陷

【1】资源分配是静态的，一个作业需要在启动时获取所需的资源并且在它的生命周期里一直持有这些资源。这导致了作业不能随负载变化而动态调整，在负载下降时无法归还空闲的资源，在负载上升时也无法动态扩展。
【2】On-Yarn模式下，所有的container都是固定大小的，导致无法根据作业需求来调整container的结构。譬如CPU密集的作业或需要更多的核，但不需要太多内存，固定结构的container会导致内存被浪费。
【3】与容器管理基础设施的交互比较笨拙，需要两个步骤来启动Flink作业：1.启动Flink守护进程；2.提交作业。如果作业被容器化并且将作业部署作为容器部署的一部分，那么将不再需要步骤2。
【4】On-Yarn模式下，作业管理页面会在作业完成后消失不可访问。
【5】Flink推荐 per job clusters 的部署方式，但是又支持可以在一个集群上运行多个作业的session模式，令人疑惑。

在Flink版本1.5中引入了Dispatcher，Dispatcher是在新设计里引入的一个新概念。Dispatcher会从Client端接受作业提交请求并代表它在集群管理器上启动作业。引入Dispatcher的原因主要有两点：
【1】一些集群管理器需要一个中心化的作业生成和监控实例；
【2】能够实现Standalone模式下JobManager的角色，且等待作业提交。在一些案例中，Dispatcher是可选的Yarn或者不兼容的kubernetes。

资源调度模型重构下的 Flink On Yarn 模式

[点击并拖拽以移动] ​

客户端提交JobGraph以及依赖jar包到YarnResourceManager，接着Yarn ResourceManager分配第一个container以此来启动AppMaster，Application Master中会启动一个FlinkResourceManager以及JobManager，JobManager会根据JobGraph生成的ExecutionGraph以及物理执行计划向FlinkResourceManager申请slot，FlinkResoourceManager会管理这些slot以及请求，如果没有可用slot就向Yarn的ResourceManager申请container，container启动以后会注册到FlinkResourceManager，最后JobManager会将subTask deploy到对应container的 slot中去。
[点击并拖拽以移动] ​

在有Dispatcher的模式下：会增加一个过程，就是Client会直接通过HTTP Server的方式，然后用Dispatcher将这个任务提交到Yarn ResourceManager中。

新框架具有四大优势，详情如下：
【1】client直接在Yarn上启动作业，而不需要先启动一个集群然后再提交作业到集群。因此client再提交作业后可以马上返回。
【2】所有的用户依赖库和配置文件都被直接放在应用的classpath，而不是用动态的用户代码classloader去加载。
【3】container在需要时才请求，不再使用时会被释放。
【4】“需要时申请”的container分配方式允许不同算子使用不同profile (CPU和内存结构)的container。

新的资源调度框架下 single cluster job on Yarn 流程介绍

[点击并拖拽以移动] ​

single cluster job on Yarn模式涉及三个实例对象：
【1】clifrontend： Invoke App code；生成StreamGraph，然后转化为JobGraph；
【2】YarnJobClusterEntrypoint（Master）： 依次启动YarnResourceManager、MinDispatcher、JobManagerRunner三者都服从分布式协同一致的策略；JobManagerRunner将JobGraph转化为ExecutionGraph，然后转化为物理执行任务Execution，然后进行deploy，deploy过程会向 YarnResourceManager请求slot，如果有直接deploy到对应的YarnTaskExecutiontor的slot里面，没有则向Yarn的ResourceManager申请，带container启动以后deploy。
【3】YarnTaskExecutorRunner (slave)： 负责接收subTask，并运行。

整个任务运行代码调用流程如下图

[点击并拖拽以移动] ​

subTask在执行时是怎么运行的？

调用StreamTask的invoke方法，执行步骤如下：
【1】initializeState()即operator的initializeState()；
【2】openAllOperators()即operator的open()方法；
【3】最后调用run方法来进行真正的任务处理；

我们来看下flatMap对应的OneInputStreamTask的run方法具体是怎么处理的。

@Override
protected void run() throws Exception {// 在堆栈上缓存处理器引用，使代码更易于JITfinal StreamInputProcessor<IN> inputProcessor = this.inputProcessor;while (running && inputProcessor.processInput()) {// 所有的工作都发生在“processInput”方法中}
}

最终是调用StreamInputProcessor的processInput()做数据的处理，这里面包含用户的处理逻辑。

public boolean processInput() throws Exception {if (isFinished) {return false;}if (numRecordsIn == null) {try {numRecordsIn = ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();} catch (Exception e) {LOG.warn("An exception occurred during the metrics setup.", e);numRecordsIn = new SimpleCounter();}}while (true) {if (currentRecordDeserializer != null) {DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);if (result.isBufferConsumed()) {currentRecordDeserializer.getCurrentBuffer().recycleBuffer();currentRecordDeserializer = null;}if (result.isFullRecord()) {StreamElement recordOrMark = deserializationDelegate.getInstance();//处理watermarkif (recordOrMark.isWatermark()) {// handle watermark//watermark处理逻辑，这里可能引起timer的triggerstatusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);continue;} else if (recordOrMark.isStreamStatus()) {// handle stream statusstatusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);continue;//处理latency watermark} else if (recordOrMark.isLatencyMarker()) {// handle latency markersynchronized (lock) {streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());}continue;} else {//用户的真正的代码逻辑// now we can do the actual processingStreamRecord<IN> record = recordOrMark.asRecord();synchronized (lock) {numRecordsIn.inc();streamOperator.setKeyContextElement1(record);//处理数据streamOperator.processElement(record);}return true;}}}//这里会进行checkpoint barrier的判断和对齐，以及不同partition 里面checkpoint barrier不一致时候的，数据buffer，final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();if (bufferOrEvent != null) {if (bufferOrEvent.isBuffer()) {currentChannel = bufferOrEvent.getChannelIndex();currentRecordDeserializer = recordDeserializers[currentChannel];currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());}else {// Event receivedfinal AbstractEvent event = bufferOrEvent.getEvent();if (event.getClass() != EndOfPartitionEvent.class) {throw new IOException("Unexpected event: " + event);}}}else {isFinished = true;if (!barrierHandler.isEmpty()) {throw new IllegalStateException("Trailing data in checkpoint barrier handler.");}return false;}}
}

streamOperator.processElement(record)最终会调用用户的代码处理逻辑，假如operator是StreamFlatMap的话。

@Override
public void processElement(StreamRecord<IN> element) throws Exception {collector.setTimestamp(element);userFunction.flatMap(element.getValue(), collector);//用户代码
}

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

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

相关文章

Docker构建缓存

Docker构建缓存

Docker镜像的分层结构 Docker的镜像是由一层一层的文件系统组成&#xff0c;以UnionFS&#xff08;联合文件系统&#xff09;堆叠构成Dockerfile中的每个指令都会创建一个新的镜像层镜像层将被缓存和复用当Dockerfile的指令修改了&#xff0c;复制的文件变化了&#xff0c;或者…
阅读更多...
Linux环境安装1

Linux环境安装1

一 概述 1.1 概要 模拟真实项目,碰到难题,使用技术/解决方案/架构设计思想;缓存架构,高并发;基于hystrix&#xff0c;缓存架构高可用的&#xff0c;高可用架构的设计以及相关的技术;商品详情页系统架构 -> 缓存架构 -> 高并发技术解决方案架构 -> 高可用技术解决方案…
阅读更多...
【c++】遍历一棵树来获取信息，并根据这些信息用map生成另一棵树，新树的键是string类型，值是char*类型

【c++】遍历一棵树来获取信息，并根据这些信息用map生成另一棵树，新树的键是string类型，值是char*类型

主要思路 递归遍历一棵树&#xff0c;将获取的信息以键值对的形式存放到c的vector容器中&#xff0c;然后遍历vector容器中的键值对信息&#xff0c;利用map容器生成个另一棵树。 具体来说&#xff0c;就是使用std::pair<std::string, const char*>类型的向量infoVector来…
阅读更多...
S7通信协议解析

S7通信协议解析

我们以S7的1500系列来查看握手和读取、写入【字Word或者位Bit】命令报文 以下报文不做说明时都是十六进制字节。 西门子PLC需要连接成功后发送两次握手命令方可进行读写通信。 西门子PLC的S7协议的头由四个字节组成.。 第一个字节数固定为03,第二个字节数固定为00 第三个字…
阅读更多...
掌握 C++ 中 static 关键字的多种使用场景

掌握 C++ 中 static 关键字的多种使用场景

static是什么 在最开始C中引入了static关键字可以用于修饰变量和函数&#xff0c;后来由于C引入了class的概念&#xff0c;现在static可以修饰的对象分为以下5种&#xff1a; 成员变量&#xff0c;成员函数&#xff0c;普通函数&#xff0c;局部变量&#xff0c; 全局变量 s…
阅读更多...
github Copilot的基本使用

github Copilot的基本使用

一.GitHub Copilot的基本介绍 GitHub Copilot 是由 GitHub 和 OpenAI 合作推出的一款代码自动补全工具&#xff0c;它基GPT&#xff08;Generative Pre-trained Transformer&#xff09;技术&#xff0c;可以为程序员提供实时的代码提示和建议。以下是 GitHub Copilot 的基本使…
阅读更多...
【leetcode】栈与队列总结

【leetcode】栈与队列总结

本文内容来自于代码随想录 栈 用栈实现队列 两个栈实现队列。思路&#xff1a;两个栈分别表示入栈和出栈。 入队&#xff1a;直接入栈出队&#xff1a; a. 出栈为空&#xff0c;先把入栈中的元素全部放到出栈中&#xff08;相当于反过来&#xff0c;这样在出栈的时候先进的元…
阅读更多...
mysql间隙锁demo分析

mysql间隙锁demo分析

概述 通常用的mysql都是innodb引擎&#xff1b; 一般在update的时候用id都会认为是给行记录加锁&#xff1b; 在使用非唯一索引更新时&#xff0c;会遇到临键锁&#xff08;范围锁&#xff09;&#xff1b; 临键锁和表中的数据有关&#xff1b; mysq版本:8 隔离级别&#xf…
阅读更多...
SpringBoot整合mail进行发送邮箱

SpringBoot整合mail进行发送邮箱

Spring Boot整合邮箱进行发送 1. 添加依赖 在pom.xml文件中添加spring-boot-starter-mail依赖&#xff0c;如下所示&#xff1a; <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-mail</artifactId>…
阅读更多...
2023年全国职业院校技能大赛网络系统管理 网络模块 出口配置

2023年全国职业院校技能大赛网络系统管理 网络模块 出口配置

（四）出口网络配置 1.北京综合服务中心办公终端可通过出口路由器R2 G 0/0.21子接口的NAPT方式访问互联网。 ip route vrf BG 0.0.0.0 0.0.0.0 GigabitEthernet 0/0.21 21.1.1.1 ip access-list extended 100 10 permit ip 10.1.20.0 0.0.0.255 any 20 permit ip 10.2.20.0 0…
阅读更多...
jmeter的常用功能及在测试中的基本使用和压测实战

jmeter的常用功能及在测试中的基本使用和压测实战

Jmeter基础功能 了解Jmeter的常用组件 元件&#xff1a;多个类似功能组件的容器&#xff08;类似于类&#xff09; 一&#xff1a;Test Plan&#xff08;测试计划&#xff09; 测试计划通常用来给测试的项目重命名&#xff0c;使用多线程脚本运行时还可以配置线程组运行方式…
阅读更多...
HCIA-Datacom题库（自己整理分类的）——STP协议判断

HCIA-Datacom题库（自己整理分类的）——STP协议判断

默认情况下&#xff0c;STP协议中根桥的根路径开销一定是0。√ 根桥交换机上所有的端口都是指定端口。 交换网络存在冗余链路时&#xff0c;使用STP可以解决交换网络中的环路问题。√ 当交换机有冗余链路时&#xff0c;使用STP可以解决问题。√ 交换机组成的网络不开启STP&…
阅读更多...
向新字符设备驱动代码框架中添加Led功能函数

向新字符设备驱动代码框架中添加Led功能函数

一. 简介 上一篇文章学习编写新字符设备驱动框架代码。文章地址如下&#xff1a; 新字符设备驱动框架代码搭建-CSDN博客 本文在以上这篇文章代码实现的基础上&#xff0c;加入涉及 Led灯的功能函数集实现。 代码实现要求&#xff1a;测试程序&#xff08;即应用程序&#…
阅读更多...
【C++】STL 容器 - map 关联容器 ① ( std::map 容器简介 | std::map 容器排序规则 | std::map 容器底层实现 )

【C++】STL 容器 - map 关联容器 ① ( std::map 容器简介 | std::map 容器排序规则 | std::map 容器底层实现 )

文章目录 一、std::map 容器1、std::map 容器简介2、std::map 容器排序规则3、std::map 容器底层实现 二、代码示例 - std::map 容器1、代码示例2、执行结果 一、std::map 容器 1、std::map 容器简介 std::map 容器 是 C 语言 标准模板库 ( STL , Standard Template Library ) …
阅读更多...
分布式技术之数据复制技术

分布式技术之数据复制技术

文章目录 什么是数据复制技术&#xff1f;数据复制技术原理及应用同步复制技术原理及应用异步复制技术原理及应用半同步复制技术原理及应用三种数据复制技术对比 什么是数据复制技术&#xff1f; 数据复制是一种实现数据备份的技术。数据复制技术&#xff0c;可以保证存储在不…
阅读更多...
Plantuml之甘特图语法介绍(二十八)

Plantuml之甘特图语法介绍(二十八)

简介&#xff1a; CSDN博客专家&#xff0c;专注Android/Linux系统&#xff0c;分享多mic语音方案、音视频、编解码等技术&#xff0c;与大家一起成长&#xff01; 优质专栏&#xff1a;Audio工程师进阶系列【原创干货持续更新中……】&#x1f680; 优质专栏&#xff1a;多媒…
阅读更多...
基于SpringBoot的在线远程考试系统

基于SpringBoot的在线远程考试系统

文章目录 项目介绍主要功能截图：部分代码展示设计总结项目获取方式🍅 作者主页：超级无敌暴龙战士塔塔开 🍅 简介：Java领域优质创作者🏆、 简历模板、学习资料、面试题库【关注我，都给你】 🍅文末获取源码联系🍅 项目介绍 基于SpringBoot的在线远程考试系统,java…
阅读更多...
磁盘和文件系统管理

磁盘和文件系统管理

一&#xff1a;磁盘结构&#xff1a; 1.磁盘基础&#xff1a; 扇区固定大小&#xff0c;每个扇区4k。磁盘会进行磨损&#xff0c;损失生命周期。 设备文件&#xff1a; 一切皆文件 设备文件&#xff1a;关联至一个设备驱动程序&#xff0c;进而能够跟与之对应硬件设备进行通…
阅读更多...
UDP发送和接受数据

UDP发送和接受数据

发送数据 public class sendmessage {public static void main (String[] args) throws IOException {DatagramSocket dsnew DatagramSocket();//打包数据开始String s"hello world";byte[] bs.getBytes();//获取InetAddress的对象InetAddress addressInetAddress.g…
阅读更多...
什么是IDE？新手用哪个IDE比较

什么是IDE？新手用哪个IDE比较

IDE代表集成开发环境&#xff08;Integrated Development Environment&#xff09;&#xff0c;它是一种软件应用程序&#xff0c;提供了一套工具&#xff0c;用于编写、调试和运行软件程序。一个IDE通常包含代码编辑器、编译器、调试器和其他各种工具&#xff0c;以便开发人员…
阅读更多...
最新文章

Copyright @ 2022~2023