听说你熟悉Flink-On-Yarn的部署模式?
5万人关注的大数据成神之路,不来了解一下吗?
5万人关注的大数据成神之路,真的不来了解一下吗?
5万人关注的大数据成神之路,确定真的不来了解一下吗?
欢迎您关注《大数据成神之路》
- 前言
Flink提供了两种在yarn上运行的模式,分别为Session-Cluster和Per-Job-Cluster模式,本文分析两种模式及启动流程。
下图展示了Flink-On-Yarn模式下涉及到的相关类图结构
- Session-Cluster模式
Session-Cluster模式需要先启动集群,然后再提交作业,接着会向yarn申请一块空间后,资源永远保持不变。如果资源满了,下一个作业就无法提交,只能等到yarn中的其中一个作业执行完成后,释放了资源,下个作业才会正常提交。所有作业共享Dispatcher和ResourceManager;共享资源;适合规模小执行时间短的作业。
2.1. 启动集群
运行bin/yarn-session.sh即可默认启动包含一个TaskManager(内存大小为1024MB,包含一个Slot)、一个JobMaster(内存大小为1024MB),当然可以通过指定参数控制集群的资源,如-n指定TaskManager个数,-s指定每个TaskManager中Slot的个数;其他配置项,可参考
下面以bin/yarn-session.sh为例,分析Session-Cluster启动流程。
2.2. 流程分析
下面分为本地和远程分析启动流程,其中本地表示在客户端的启动流程,远端则表示通过Yarn拉起Container的流程;
2.2.1 本地流程
Session启动入口为FlinkYarnSessionCli#main
根据传入的参数确定集群的资源信息(如多少个TaskManager,Slot等)
部署集群AbstractYarnClusterDescriptor#deploySessionCluster -> AbstractYarnClusterDescriptor#deployInternal
根据flink-conf的HA配置创建对应的服务(如StandaloneHaServices、ZooKeeperHaServices等)
创建基于Netty的RestClient;
创建/rest_server_lock、/dispatcher_lock节点(以ZK为例)
启动监听节点的变化(主备切换)
初始化文件系统(HDFS)
将log4j、logback、flink-conf.yaml、jar包上传至HDFS
构造AppMaster的Container(确定Container进程的入口类YarnSessionClusterEntrypoint),构造相应的Env
YarnClient向Yarn提交Container申请
跟踪ApplicationReport状态(确定是否启动成功,可能会由于资源不够,一直等待)
进行资源校验(如内存大小、vcore大小、队列)
通过YarnClient创建Application
再次校验资源
AbstractYarnClusterDescriptor#startAppMaster启动AppMaster
启动成功后将对应的ip和port写入flinkConfiguration中
创建与将集群交互的ClusterClient
通过ClusterClient获取到appId信息并写入本地临时文件
经过上述步骤,整个客户端的启动流程就结束了,下面分析yarn拉起Session集群的流程,入口类在申请Container时指定为YarnSessionClusterEntrypoint。
2.2.2 远端流程
远端宿主在Container中的集群入口为YarnSessionClusterEntrypoint#main
ClusterEntrypoint #runClusterEntrypoint -> ClusterEntrypoint#startCluster启动集群
经过上述步骤就完成了集群的启动;
创建/dispatcher_lock节点,/resource_manager_lock节点
创建DispatcherGateway、ResourceManagerGateway的Retriever(用于创建RpcGateway)
创建DispatcherGateway的WebMonitorEndpoint并启动
创建JobManager的指标组
创建ResourceManager、Dispatcher并启动进行ZK选举
返回SessionDispatcherResourceManagerComponent
初始化文件系统
初始化各种Service(如:
创建RpcService(AkkaRpcService)、创建HAService、创建并启动BlobServer、创建HeartbeatServices、创建指标服务并启动、创建本地存储ExecutionGraph的Store)
创建DispatcherResourceManagerComponentFactory(SessionDispatcherResourceManagerComponentFactory),用于创建DispatcherResourceManagerComponent(用于启动Dispatcher、ResourceManager、WebMonitorEndpoint)
通过DispatcherResourceManagerComponentFactory创建DispatcherResourceManagerComponent
2.3. 启动任务
当启动集群后,即可使用./flink run -c mainClass /path/to/user/jar向集群提交任务。
2.4 流程分析
同样,下面分为本地和远程分析启动流程,其中本地表示在客户端提交任务流程,远端则表示集群收到任务后的处理流程。
2.4.1 本地流程
程序入口为CliFrontend#main
解析处理参数
根据用户jar、main、程序参数、savepoint信息生成PackagedProgram
获取session集群信息
执行用户程序
获取任务的JobGraph
通过RestClusterClient#submitJob提交任务
创建本地临时文件存储JobGraph
通过RestClusterClient向集群的rest接口提交任务
处理请求响应结果
设置ClassLoader
设置Context
执行用户程序main方法(当执行用户业务逻辑代码时,会解析出StreamGraph然后通过ClusterClient#run来提交任务),其流程如下:
重置Context
重置ClassLoader
经过上述步骤,客户端提交任务过程就完成了,主要就是通过RestClusterClient将用户程序的JobGraph通过Rest接口提交至集群中。
2.4.2 远端流程
远端响应任务提交请求的是RestServerEndpoint,其包含了多个Handler,其中JobSubmitHandler用来处理任务提交的请求;
处理请求入口:
JobSubmitHandler#handleRequest
进行相关校验
从文件中读取出JobGraph
通过BlobClient将jar及JobGraph文件上传至BlobServer中
通过Dispatcher#submitJob提交JobGraph
通过Dispatcher#runJob运行任务
启动JobMaster
将任务运行状态信息写入ZK(/${AppID}/running_job_registry/${jobId})
启动JobMaster的Endpoint
开始调度任务JobMaster#startJobExecution
创建JobManagerRunner(处理leader选举)
创建JobMaster(实际执行任务入口,包含在JobManagerRunner)
启动JobManagerRunner(会进行leader选举,ZK目录为leader/${jobId}/job_manager_lock)
当为主时会调用JobManagerRunner#grantLeadership方法
接下来就进行任务具体调度(构造ExecutionGraph、申请Slot等)流程,本篇文章不再展开介绍。
- Per-Job-Cluster模式
一个任务会对应一个Job,每提交一个作业会根据自身的情况,都会单独向yarn申请资源,直到作业执行完成,一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager,按需接受资源申请;适合规模大长时间运行的作业。
3.1 启动任务
启动Per-Job-Cluster任务,可通过./bin/flink run -m yarn-cluster -d -c mainClass /path/to/user/jar命令使用分离模式启动一个集群,即单任务单集群;
3.2. 流程分析
与Session-Cluster类似,我们对Per-Job-Cluster模式也分为本地和远端。
3.2.1 本地流程
与Session-Cluster模式类似,入口也为CliFrontend#main
解析处理参数
根据用户jar、main、程序参数、savepoint信息生成PackagedProgram
根据PackagedProgram创建JobGraph(对于非分离模式还是和Session模式一样,模式Session-Cluster)
获取集群资源信息
部署集群YarnClusterDesriptor#deployJobCluster -> AbstractYarnClusterDescriptor#deployInternal;
后面流程与Session-Cluster类似,值得注意的是在AbstractYarnClusterDescriptor#startAppMaster中与Session-Cluster有一个显著不同的就是其会将任务的JobGraph上传至Hdfs供后续服务端使用
经过上述步骤,客户端提交任务过程就完成了,主要涉及到文件(JobGraph和jar包)的上传。
3.2.2 远端流程
远端宿主在Container中的集群入口为YarnJobClusterEntrypoint#main
ClusterEntrypoint#runClusterEntrypoint -> ClusterEntrypoint#startCluster启动集群
创建JobDispatcherResourceManagerComponentFactory(用于创建JobDispatcherResourceManagerComponent)
创建ResourceManager(YarnResourceManager)、Dispatcher(MiniDispatcher),其中在创建MiniDispatcher时会从之前的JobGraph文件中读取出JobGraph,并启动进行ZK选举
当为主时会调用Dispatcher#grantLeadership方法
Dispatcher#runJob开始运行任务(创建JobManagerRunner并启动进行ZK选举),后续流程与Session-Cluster相同,不再赘述
Dispatcher#recoverJobs恢复任务,获取JobGraph
Dispatcher#tryAcceptLeadershipAndRunJobs确认获取主并开始运行任务
Flink提供在Yarn上两种运行模式:Session-Cluster和Per-Job-Cluster,其中Session-Cluster的资源在启动集群时就定义完成,后续所有作业的提交都共享该资源,作业可能会互相影响,因此比较适合小规模短时间运行的作业,对于Per-Job-Cluster而言,所有作业的提交都是单独的集群,作业之间的运行不受影响(可能会共享CPU计算资源),因此比较适合大规模长时间运行的作业。
5万人关注的大数据成神之路,不来了解一下吗?
5万人关注的大数据成神之路,真的不来了解一下吗?
5万人关注的大数据成神之路,确定真的不来了解一下吗?
欢迎您关注《大数据成神之路》
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Spark中几种ShuffleWriter的区别你都知道吗?
一.前言 在Spark中有三种shuffle写,分别是BypassMergeSortShuffleWriter、UnsafeShuffleWriter、SortShuffleWriter。分别对应三种不同的shuffleHandle。 这三者和ShuffleHandle的对应关系如下: UnsafeShuffleWriter:SerializedShuffleHandle BypassMergeSortShuffleWriter:BypassMergeSortShuffleHandle, SortShuffleWriter:BaseShuffleHandle 那么这些shuffle写内部的实现细节有何不同,在什么场景下使用什么样的shuffleWriter呢,接下来我们对着三种ShuffleWriter的实现细节做一个比较。 二.不同shuffleHandle的使用时机 不同shuffleWrite的使用其实是根据shuffleHandle来决定的,在构建shuffleDependence时都会构建shuffleHandle,在registerShuffle方法中,有着对shuffle...
- 下一篇
你有必要了解一下Flink底层RPC使用的框架和原理
5万人关注的大数据成神之路,不来了解一下吗?5万人关注的大数据成神之路,真的不来了解一下吗?5万人关注的大数据成神之路,确定真的不来了解一下吗? 欢迎您关注《大数据成神之路》 前言 对于Flink中各个组件(JobMaster、TaskManager、Dispatcher等),其底层RPC框架基于Akka实现,本文着重分析Flink中的Rpc框架实现机制及梳理其通信流程。 Akka介绍 由于Flink底层Rpc是基于Akka实现,我们先了解下Akka的基本使用。 Akka是一个开发并发、容错和可伸缩应用的框架。它是Actor Model的一个实现,和Erlang的并发模型很像。在Actor模型中,所有的实体被认为是独立的actors。actors和其他actors通过发送异步消息通信。Actor模型的强大来自于异步。它也可以显式等待响应,这使得可以执行同步操作。但是,强烈不建议同步消息,因为它们限制了系统的伸缩性。每个actor有一个邮箱(mailbox),它收到的消息存储在里面。另外,每一个actor维护自身单独的状态。一个Actors网络如下所示: 每个actor是一个单一的线程,...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS6,CentOS7官方镜像安装Oracle11G
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- CentOS关闭SELinux安全模块
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- Windows10,CentOS7,CentOS8安装MongoDB4.0.16
- Hadoop3单机部署,实现最简伪集群
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- Eclipse初始化配置,告别卡顿、闪退、编译时间过长
- SpringBoot2全家桶,快速入门学习开发网站教程