Flink内部原理

部署、停止、取消Task触发检查点（ Deploy、StopCancel TasksTrigger Checkpoints）

3. 2 AM启动Dispatcher

内存和IO管理器(Memory & T/O Manager)

Flink程序(Flink Program)

接着1.5会进入AbstractJobClusterExecutor，等同于YarnJobClusterExecutor1.6 ①生成JobGraph，也就是getJobGraph方法将 流图（StreamGraph） 转换成 作业图（JobGraph）②创建并启动yarn客户端，font color=\"#ff0000\

TaskSlot

Task

JobManager

Actor通信系统(Actor System)

ApplicationMaster

11. 提交执行sumitTask()

10. 提供slot

YarnJobClusterExecutor

CliFrontend

3. 1 AM启动ResourceManager

4 注册、请求slot使用的是SlotPoolImpl#requestSlotFromResourceManage方法发送请求

5.申请资源通过ActiveResourceManager#startNewWorker方法去申请

7. runTaskManager启动TaskExecutor

Task状态心跳监控统计（Task StatusHeartbeatsStatistics）

bin/flink run -t yarn-per-job -c com.xxx.xxx.WordCount./WordCount.jar 

6.启动TaskManager

NodeManager

3.4 生成ExecutionGraph

8. 注册slot

通过工厂类创建Dispatcher创建对象会涉及到start和onStart1.接收用户的作业，2.启动JobMaster

网络管理器(Network Manager)

YarnClientYarnClusterClientFactory#getClusterDescriptor里面创建了YarnClient，这个客户端是用来和Yarn做交互的

YarnTaskExecutorRunner

1.9 提交任务信息submitApplication

SlotPool

3.3 Dispatcher启动JobMaster

                                                                                                         找到CliFrontend#main方法1.2 CliFrontendParser对解析参数1.3 选择创建哪种类型的客户端？--> Generic、Yarn、Default三种模式，后面代码按照：①是否指定为per-job模式，即指定”-m yarn-cluster”; ID = \"yarn-cluster\"②是否存在flink在yarn的appID，即yarn-session模式是否启动③executor的名字为 \"yarn-session\" 或 \"yarn-per-job\"我们是yarn-per-job模式，所以最后会选择FlinkYarnSessionCli模式使用FlinkYarnSessionCli1.4 CliFrontend#getJobJarAndDependencies加载我们用户的Jar包，PackagedProgram#callMainMethod执行用户代码，找到这段代码StreamExecutionEnvironment#execute()进入这个方法，span style=\"font-size: inherit;\

数据流

TaskManager

提交任务、发送数据流、取消或更新任务（Submit job、Send Dataflow、Cancel、Update Job）

9. 分配slot

HDFS

状态更新、计算结果（Status updatesStatistics & results）

SlotManager实时检查所需slot是否足够步骤9通过SlotManagerImpl#allocateSlot通知TM提供slot给 JM

                                               在Yarn模式下的进程名称叫YarnJobClusterEntrypointPer-job模式的ApplicationMaster Container加载运行入口是YarnJobClusterEntryPoint中的main()方法，这个方法里面ClusterEntrypoint.runClusterEntrypoint运行集群的入口：①初始化服务：Rpc相关，设置JobManager的地址与端口，创建Dispatcher、ResourceManager对象的工厂类②通过工厂类创建和启动Dispatcher、ResourceManager对象，③Dispatcher创建启动的同时也会创建启动JobMaster（里面有一个slotpool，真正去发送请求的)    注意：在创建JobMaster的时候是直接new JobMaster，这个是一个很大的构造器，这个构造器里面有：创建调度器，同时创建的时候把 JobGraph转换成ExecutionGraph，ExecutionGraph的入口就在这边④ResourceManager的创建，启动RM的时候还会顺带去创建了Yarn的RM和NM的客户端，初始化并启动，里面有一个slotmanager（真正管理资源的、向yarn申请资源）    注意：在创建RM的时候有关很关键的东西就是YarnResourceManagerDriver#initializeInternal这个方法很重要会对接到TaskManager启动那边⑤slotpool向slotmanager申请资源，slotmanager向yarn申请资源，启动新节点

YarnClusterDescriptor（集群描述器）1.7 上传jar包和配置1.8 封装提交参数和命令bin/java ApplicationMaster--jar … --class …

作业调度器(Scheduler)

JobMaster

检查点协调器(Checkpoint Coordinator)

1.1 脚本启动执行

Yarn的ResourceManager（资源管理）

数据流图(Dataflow Graph)