flink提交流程源码

flink源码系列总述

本文基于flink-1.17版本，对于flink源码学习了解，仅作为个人学习笔记，如有错误，欢迎指正。

flink提交流程源码 流程解析

看以下流程时，请及时参考本图

1. CliFrontend 客户端
2. YarnJobClusterEntrypoint AM执行的入口
3. YarnTaskExecutorRunner Yarn模式下TaskManager的入口类

1. CliFrontend 客户端

提交命令
通过flink on yarn per-job模式提交，查看flink脚本可以看到，程序被提交后，会寻找CliFrontend类

CliFrotend n main方法入口

其中： parseAndRun方法

其中：parseAndRun的run中：

获取有效配置中：

执行程序中：

执行用户代码：
env.execute() ->StreamExecutionEnvironment.execute()

获取StreamGraph StreamExecutionEnvironment类

获取PipelineExecutor执行器 执行

生成jobGraph YarnJobClusterExecutor类

其中1：
创建yarnClient 生成YarnClusterDescriptor YarnClusterClientFactory类

其中2：
启动am ApplicationMaster

2.在部署集群的时候，已经启动了ApplicationMaster

3.ApplicationMaster 执行过程

YarnJobClusterEntrypoint 启动入口 main方法

工厂类构建和对象创建

对象创建：
创建resourceManager

创建并启动dispatcher 启动resourceManager

其中：创建并启动dispatcher

1. DefaultDispatcherRunnerFactory -> createDispatcherRunner
2. DefaultDispatcherRunner -> create
3. DefaultDispatcherRunner -> start
4. StandaloneLeaderElectionService -> start
5. DefaultDispatcherRunner -> grantLeadership
6. DefaultDispatcherRunner -> startNewDispatcherLeaderProcess
7. AbstractDispatcherLeaderProcess -> start
8. AbstractDispatcherLeaderProcess -> startInternal
9. JobDispatcherLeaderProcess -> onStart
10. DefaultDispatcherGatewayServiceFactory -> create
11. DefaultDispatcherGatewayServiceFactory
12. Dispatcher -> onStart

其中：dispatcher服务启动主要进行注册
其中：启动jobMaster

1. Dispatcher -> startCleanupRetries
2. Dispatcher -> runCleanupRetry
3. Dispatcher -> runJob
4. JobMasterServiceLeadershipRunner -> start
5. StandaloneLeaderElectionService -> start
6. JobMasterServiceLeadershipRunner -> grantLeadership
7. JobMasterServiceLeadershipRunner -> startJobMasterServiceProcessAsync
8. JobMasterServiceLeadershipRunner -> verifyJobSchedulingStatusAndCreateJobMasterServiceProcess
9. JobMasterServiceLeadershipRunner -> createNewJobMasterServiceProcess
10. DefaultJobMasterServiceProcessFactory -> create
11. DefaultJobMasterServiceProcess -> DefaultJobMasterServiceProcess
12. DefaultJobMasterServiceFactory -> createJobMasterService
13. DefaultJobMasterServiceFactory -> internalCreateJobMasterService
14. 启动

其中： 启动resourceManager

1. ResourceManagerServiceImpl -> grantLeadership
2. ResourceManagerServiceImpl -> startNewLeaderResourceManager
3. ResourceManagerServiceImpl -> startResourceManagerIfIsLeader
4. resourceManager -> start
5. ResourceManager -> onStart
6. ResourceManager -> startResourceManagerServices

initializate():
创建yarn的reasouceManager和nodeManager的client 并启动

其中：启动jobMaster
JobMaster -> onStart

真正启动jobMaster

建立连接，开始请求资源

1. StandaloneLeaderRetrievalService -> start
2. JobMaster -> notifyOfNewResourceManagerLeader
3. JobMaster -> reconnectToResourceManager
4. JobMaster -> tryConnectToResourceManager
5. JobMaster -> connectToResourceManager
6. RegisteredRpcConnection -> start

注册成功调用 onRegistrationSuccess(), 向 ResourceManager 进行 slot 的申请

1. JobMaster -> onRegistrationSuccess
2. JobMaster -> establishResourceManagerConnection
3. DeclarativeSlotPoolService -> connectToResourceManager
4. ResourceManager -> declareRequiredResources
5. FineGrainedSlotManager -> processResourceRequirements

其中：notifyResourceRequirements

1. DefaultResourceTracker -> notifyResourceRequirements
2. JobScopedResourceTracker -> notifyResourceRequirements

其中：checkResourceRequirementsWithDelay

1. FineGrainedSlotManager -> checkResourceRequirementsWithDelay
2. FineGrainedSlotManager -> checkResourceRequirements

其中 tryFulfillRequirements

tryFulfilledRequirementWithResource

其中：分配资源allocateSlotsAccordingTo

其中：declareNeededResourcesWithDelay

1. FineGrainedSlotManager -> declareNeededResourcesWithDelay
2. FineGrainedSlotManager -> declareNeededResources
3. ActiveResourceManager -> declareResourceNeeded
4. ActiveResourceManager -> declareResourceNeeded
5. ActiveResourceManager -> checkResourceDeclarations
6. ActiveResourceManager -> requestNewWorker
7. YarnResourceManagerDriver -> requestResource
8. YarnResourceManagerDriver -> addContainerRequest

请求容器 进行回调函数YarnResourceManagerDriver ->onContainersAllocated
2. YarnResourceManagerDriver ->onContainersOfPriorityAllocated

1. YarnResourceManagerDriver -> startTaskExecutorInContainerAsync
2. YarnResourceManagerDriver -> createTaskExecutorLaunchContext
3. Utils.createTaskExecutorContext

所以 能够 启动TaskExecutorRunner main方法

1. TaskExecutorRunner -> runTaskManagerProcessSecurely
2. TaskExecutorRunner -> runTaskManagerProcessSecurely
3. TaskExecutorRunner -> runTaskManager
4. TaskExecutorToServiceAdapter -> start

TaskExecutor onstart方法：

1. TaskExecutor -> onStart
2. TaskExecutor -> startTaskExecutorServices

其中resourceManagerLeaderRetriever.start

1. ResourceManagerLeaderListener -> notifyLeaderAddress
2. TaskExecutor -> notifyOfNewResourceManagerLeader
3. TaskExecutor -> reconnectToResourceManager
4. TaskExecutor -> tryConnectToResourceManager
5. TaskExecutor -> connectToResourceManager

其中：newRegistration.startRegistration(); // todo invokeRegistration

1. RetryingRegistration -> startRegistration
2. RetryingRegistration -> register
3. RetryingRegistration -> invokeOperation
4. TaskExecutorToResourceManagerConnection -> invokeRegistration
5. registerTaskExecutor registerTaskExecutorInternal

注册成功调用 onRegistrationSuccess(), 向 ResourceManager 进行 slot 的申请

1. TaskExecutorToResourceManagerConnection -> onRegistrationSuccess
2. ResourceManagerRegistrationListener -> onRegistrationSuccess
3. ResourceManagerRegistrationListener -> establishResourceManagerConnection
4. ResourceManager -> sendSlotReport
5. FineGrainedSlotManager -> registerTaskManager 进行allocateSlot

1. DefaultSlotStatusSyncer -> allocateSlot
2. TaskExecutor(!!!) -> requestSlot
3. TaskExecutor -> requestSlot

其中 allocateSlotForJob:

1. TaskExecutor -> allocateSlotForJob
2. TaskExecutor -> allocateSlot
3. TaskSlotTableImpl -> allocateSlot

其中：offerSlotsToJobManager

1. TaskExecutor -> internalOfferSlotsToJobManager
2. JobMaster -> offerSlots
3. DeclarativeSlotPoolService -> offerSlots
4. TestingDeclarativeSlotPool -> offerSlots
5. TestingDeclarativeSlotPool -> internalOfferSlots