Flink源码分析之Flink on YARN - Per Job

./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar

1. bin/flink run
  
2. CliFrontend.java
 	 	main(String[] args)
  		CliFrontend cli = new CliFrontend(configuration, customCommandLines)
  		retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args))
  			run(params)
  				executeProgram(Configuration, PackagedProgram)
  					ClientUtils.executeProgram
   ClientUtils.java
   	executeProgram
  		program.invokeInteractiveModeForExecution
3.  			callMainMethod(entryClass, args)
  					entryClass.getMethod("main", String[].class).invoke(null, (Object) args)
  
4. 用户程序入口类
   	main(args)
  
5. StreamExecutionEnvironment.java
   	executeAsync(StreamGraph)
  		executorFactory.getExecutor(configuration).execute(streamGraph, configuration, userClassloader)
  
6. AbstractJobClusterExecutor.java
   	execute
			{
				JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration)
        clusterDescriptor = clusterClientFactory.createClusterDescriptor(configuration)
        clusterSpecification = clusterClientFactory.getClusterSpecification(configuration)
        clusterDescriptor.deployJobCluster(clusterSpecification, jobGraph, detached)
			}
    	
7. YarnClusterDescriptor.java
   	ClusterClientProvider<ApplicationId> deployJobCluster(ClusterSpecification, JobGraph, detached)
  		deployInternal
				{		
  				1) 检查用户权限(Kerberos证书)、vcores、hadoop环境变量、yarn queue等
            
  				2) ApplicationReport report = startAppMaster(...)
          	{
              1. 上传资源到HDFS(yarnShipFiles、yarn.ship-archives、pipeline.jars、配置文件等)
              2. 将JobGraph序列化到文件中，并上传到HDFS(per job模式才有)
              3. 设置集群HA信息
              4. 设置ApplicationSubmissionContext
                    setApplicationName
                    setApplicationType
                    setAMContainerSpec    // Application Master Container
                        setCommands       // 通过class参数指定了yarn AM container启动类为YarnJobClusterEntrypoint
                        setTokens
                        setLocalResources(job.graph、flink-conf.yaml、yarn-site.xml、krb5.conf、keytab、kerberos)
                        setEnvironment
                            flink-conf.yaml中用户配置的以containerized.master.env.为前缀的变量
                            _FLINK_CLASSPATH // Flink app class path
                            _FLINK_DIST_JAR  // local path of flink dist jar 
                            classpath from YARN configuration
                            ...
                    setResource(masterMemoryMB、yarn.appmaster.vcores)
                    setPriority
                    setQueue	// yarn queue
                    setApplicationNodeLabel
                    setApplicationTags
                5. 提交yarn application： yarnClient.submitApplication(appContext)
                   // RUNNING或FINISHED状态时，正常退出循环；FAILED或KILLED时抛异常退出；状态变化时打印日志，运行超60秒打印日志
                6. loop循环等待提交结果: ApplicationReport report = yarnClient.getApplicationReport(appId)
        			}
  					3) 从ApplicationReport中获取rpc和rest的地址和端口信息以及ApplicationId信息，设置回Configuration；设置ClusterId=ApplicationId信息到HA中。
            4) 返回RestClusterClient(Configuration, ApplicationId)
				}	

YarnJobClusterEntrypoint.java
  main(String[] args)
  	ClusterEntrypoint.runClusterEntrypoint(yarnJobClusterEntrypoint)
  		clusterEntrypoint.startCluster()
  			runCluster(Configuration, PluginManager)
  				initializeServices
						{
1.            commonRpcService = AkkaRpcServiceUtils.createRemoteRpcService
              JMXService
              IO线程池
              HA Service 
              BlobServer
              HeartbeatServices
              MetricRegistry
              MetricQuery	RpcService
              ProcessMetricGroup
              ExecutionGraphInfoStore
						}
  				clusterComponent = dispatcherResourceManagerComponentFactory.create
            
DefaultDispatcherResourceManagerComponentFactory.java
	create(...)
  	{
2.   	webMonitorEndpoint = restEndpointFactory.createRestEndpoint
      webMonitorEndpoint.start()
        
3.		resourceManagerService = ResourceManagerServiceImpl.create    
  
4.    dispatcherRunner = dispatcherRunnerFactory.createDispatcherRunner
				{
  					...
            Dispatcher dispatcher = dispatcherFactory.createDispatcher
            dispatcher.start() 
            	{
              	...
5.               startRecoveredJobs()
                	{
                  	for (JobGraph recoveredJob : recoveredJobs){
                      ...
                      JobMaster jobMaster = new JobMaster(...) 
                      jobMaster.start()
                    }
                	}
            	}
				}
      resourceManagerService.start()
      	{
        	...
           YarnResourceManagerDriver.java
            initializeInternal()
          		{
            		 ...
                 // 创建并启动AMRMClientAsync，联系YARN RM
              	 resourceManagerClient = yarnResourceManagerClientFactory.createResourceManagerClient
                 resourceManagerClient.init(yarnConfig)
                 resourceManagerClient.start()
                 
                 // AM启动后向YARN RM注册自己，这样可以通过YARN跳转到Flink web ui页面
                 resourceManagerClient.registerApplicationMaster(rpcAddress, restPort, webInterfaceUrl)               
                ...
                 // 创建并启动NMClientAsync，用于联系YARN NM
                 nodeManagerClient = yarnNodeManagerClientFactory.createNodeManagerClient
                 nodeManagerClient.init(yarnConfig)
                 nodeManagerClient.start() 
          		}
           	YarnContainerEventHandler.onContainersAllocated(List<Container> containers)
          		{
							 	...
                 // 运行任务所需的Container都申请并领取完毕后，AM维护与YARN RM心跳
                 // resourceManagerClient = AMRMClientAsync
                 if (getNumRequestedNotAllocatedWorkers() <= 0) {
                 		resourceManagerClient.setHeartbeatInterval(yarnHeartbeatIntervalMillis);
                 }
          		}
      	}  
  	}
  

JobMaster.java
	new JobMaster	
  	{
    	...
      // jobmanager.scheduler 默认值为Ng，因此创建的SchedulerNG为DefaultScheduler
      this.schedulerNG = createScheduler(...)  
      	{
        	// JobGraph -> ExecutionGraph
          this.executionGraph = createAndRestoreExecutionGraph(...)  
        }
    }
	jobMaster.start 
  ...
  onStart
  	startJobExecution
    	startJobMasterServices
      startScheduling
1.    	schedulerNG.startScheduling
    
DefaultScheduler.java
	// 方法体在父类SchedulerBase.java中  
  startScheduling		
  	startSchedulingInternal
    	schedulingStrategy.startScheduling
    
PipelinedRegionSchedulingStrategy.java
	startScheduling
  	maybeScheduleRegions(Set<SchedulingPipelinedRegion> regions) 
    	// 遍历regions，按region调度
      maybeScheduleRegion
      	schedulerOperations.allocateSlotsAndDeploy(List<ExecutionVertexDeploymentOption>)
    
DefaultScheduler.java  
	allocateSlotsAndDeploy
  	allocateSlots
    	executionSlotAllocator.allocateSlotsFor(List<ExecutionVertexID> executionVertexIds)

SlotSharingExecutionSlotAllocator.java
	allocateSlotsFor
  	getOrAllocateSharedSlot
    	slotProvider.allocatePhysicalSlot(PhysicalSlotRequest)

PhysicalSlotProviderImpl.java
	allocatePhysicalSlot
  	// 先看有可用的slot没，有的话直接分配
    tryAllocateFromAvailable
    // 没的话请求Flink RM获取
    orElseGet(requestNewSlot(...))
2.   	slotPool.requestNewAllocatedSlot(SlotRequestId, ResourceProfile, timeout)
    
DeclarativeSlotPoolBridge.java
	requestNewAllocatedSlot
  	internalRequestNewSlot
    	internalRequestNewAllocatedSlot
      	getDeclarativeSlotPool().increaseResourceRequirementsBy
    
DefaultDeclarativeSlotPool.java
	increaseResourceRequirementsBy
  	declareResourceRequirements
    	notifyNewResourceRequirements.accept(Collection<ResourceRequirement>)
    
DeclarativeSlotPoolService.java  
	declareResourceRequirements(Collection<ResourceRequirement>)
  	resourceRequirementServiceConnectionManager.declareResourceRequirements
    
DefaultDeclareResourceRequirementServiceConnectionManager.java
	declareResourceRequirements(ResourceRequirements)
  	triggerResourceRequirementsSubmission
    	sendResourceRequirements
      	service.declareResourceRequirements
    
DeclarativeSlotPoolService.java
	// 向Flink RM发送RPC请求，获取slots资源
  resourceManagerGateway.declareRequiredResources(JobMasterId, ResourceRequirements, Time timeout)
 

// Flink ResourceManager服务端处理资源请求
ResourceManager.java    
	declareRequiredResources   
3.	slotManager.processResourceRequirements

DeclarativeSlotManager.java 
	processResourceRequirements
  	resourceTracker.notifyResourceRequirements
    checkResourceRequirements()
    	tryFulfillRequirementsWithPendingSlots(JobID jobId, Collection<Map.Entry<ResourceProfile, Integer>> missingResources, ResourceCounter pendingSlots)
    		// 遍历missingResource
    		tryAllocateWorkerAndReserveSlot(ResourceProfile profile, ResourceCounter pendingSlots)
    			taskExecutorManager.allocateWorker(profile)
    
TaskExecutorManager.java
	allocateWorker
  	resourceActions.allocateResource(WorkerResourceSpec)

ResourceManager.java
	ResourceActionsImpl.allocateResource
  	startNewWorker

ActiveResourceManager.java
	startNewWorker
  	requestNewWorker        
    	resourceManagerDriver.requestResource(TaskExecutorProcessSpec)

YarnResourceManagerDriver.java
	requestResource
  		// 请求获取container资源
3.1 	resourceManagerClient.addContainerRequest(getContainerRequest(resource, priority))

YarnResourceManagerDriver.java
	// container里有NM的地址信息
  YarnContainerEventHandler.onContainersAllocated(List<Container> containers)
  	onContainersOfPriorityAllocated
    	// 遍历containers
      startTaskExecutorInContainerAsync
      	// 创建ContainerLaunchContext请求对象
        context = createTaskExecutorLaunchContext(ResourceID containerId, String host, TaskExecutorProcessSpec taskExecutorProcessSpec)
        // 通过YARN NM Client发送请求，启动container运行TaskManager进程
3.2     nodeManagerClient.startContainerAsync(container, context) 

DefaultScheduler.java
  allocateSlotsAndDeploy
  	// 请求获取slots资源
  	allocateSlots
  	// 部署Task
  	waitForAllSlotsAndDeploy
  		deployAll(List<DeploymentHandle> deploymentHandles)
  			// 遍历deploymentHandles
  			deployOrHandleError(DeploymentHandle)
  				deployTaskSafe
  					executionVertexOperations.deploy(ExecutionVertex)
  
DefaultExecutionVertexOperations.java
  deploy(ExecutionVertex executionVertex)
  	executionVertex.deploy()
  
ExecutionVertex.java
  deploy()
  	currentExecution.deploy()

Execution.java
  deploy()
  	TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway()
  	taskManagerGateway.submitTask(TaskDeploymentDescriptor, rpcTimeout)
  
RpcTaskManagerGateway.java
  submitTask
  	taskExecutorGateway.submitTask(TaskDeploymentDescriptor, jobMasterId, timeout)
  
// RPC服务端响应  
TaskExecutor.java
  submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, Time timeout)
		{
  		...
      // Task implements Runnable
  		Task task = new Task(...)
      taskSlotTable.addTask(task)
      task.startTaskThread()
        // executingThread = new Thread(TASK_THREADS_GROUP, this, taskNameWithSubtask)
        executingThread.start()
		}

Task.java
	run()
    doRun()
			{
  			...
         TaskInvokable invokable = loadAndInstantiateInvokable(...)
         	{
						...
             invokableClass = Class.forName(className, true, classLoader).asSubclass(TaskInvokable.class)
             invokableClass.getConstructor(Environment.class).newInstance(environment)
        	}
  				restoreAndInvoke(invokable)
          ...
			}

// 1) 启动leader选举服务，选举leader
// start(LeaderContender) -> leaderContender.grantLeadership
{
	// implements ResourceManagerService, LeaderContender
  ResourceManagerServiceImpl.java
	  start()
  		leaderElectionService.start(this)
    
	// HA leader选举  
  DefaultLeaderElectionService.java
  	start(LeaderContender)
    	leaderElectionDriver = leaderElectionDriverFactory.createLeaderElectionDriver

  ZooKeeperLeaderElectionDriverFactory.java
  	createLeaderElectionDriver
    	new ZooKeeperLeaderElectionDriver
        	{
          	// latchPath节点用于leader选举
          	leaderLatch = new LeaderLatch(client, checkNotNull(latchPath))
            leaderLatch.addListener(this)
            leaderLatch.start()

            // leaderPath节点用于存储leader信息，监听该节点数据的变化
            cache = new NodeCache(client, leaderPath)
            cache.getListenable().addListener(this)
            cache.start()
          }

	ZooKeeperLeaderElectionDriver.java
  	// 成为leader时，latchPath上挂的监听会被回调，isLeader方法被执行
    isLeader()
    	leaderElectionEventHandler.onGrantLeadership

	DefaultLeaderElectionService.java
  	onGrantLeadership
    	leaderContender.grantLeadership   
}

// 2) 成为leader后，创建ResourceManager并启动RPC服务
{
  ResourceManagerServiceImpl.java
    grantLeadership
      startNewLeaderResourceManager
    		// resourceManagerFactory = ActiveResourceManagerFactory
				this.leaderResourceManager = resourceManagerFactory.createResourceManager 
  				{
            new ActiveResourceManager
              {
                ...
                // 启动RPC服务
                this.rpcServer = rpcService.startServer(this)
              }
  				}
    		startResourceManagerIfIsLeader
    			resourceManager.start()
}

// 3) start() -> onStart()
{
	// ResourceManager extends RpcEndpoint
  RpcEndpoint.java
  	start()
    	rpcServer.start()   

  // AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint, RpcServer
  AkkaInvocationHandler.java（客户端）
  	// rpcServer是JDK Proxy生成的代理对象，实现了InvocationHandler接口的invoke方法，故start方法交由代理对象的invoke执行
    invoke(Object proxy, Method method, Object[] args) 
    	// if(非rpc方法) 调用对象自身相应的method处理
      result = method.invoke(this, args)
      	start()
        	// rpcEndpoint是Actor的引用：ActorRef，可以用来向Actor发消息，这里是向自身ResourceManager发送控制类消息
          rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender())

	// AkkaRpcActor extends AbstractActor，是一个Actor  
  AkkaRpcActor.java（服务端）  
  	handleControlMessage(ControlMessages)
    	// 初始状态为STOPPED，即state = AkkaRpcActor.StoppedState
      state = state.start(this)
      	akkaRpcActor.rpcEndpoint.internalCallOnStart()

  RpcEndpoint.java
  	internalCallOnStart()
    	onStart()  
}

// 4) 启动ResourceManager下相关服务：心跳管理服务、SlotManager等
{
	ResourceManager.java
    onStart()
    	startResourceManagerServices()
    		...
    		startHeartbeatServices()
    		slotManager.start
}

// 入口 dispatcherRunner = dispatcherRunnerFactory.createDispatcherRunner

// 1) 创建DispatcherRunner，启动HA leader选举
{
  DefaultDispatcherRunnerFactory.java
    createDispatcherRunner
      DefaultDispatcherRunner.create
        DispatcherRunnerLeaderElectionLifecycleManager.createFor
          new DispatcherRunnerLeaderElectionLifecycleManager
    				// 启动leader选举服务， 
    				// 后续 start(LeaderContender) -> leaderContender.grantLeadership过程同上述ResourceManager调用链一致，不再多述
            leaderElectionService.start(dispatcherRunner)  
}

// 2) 成为leader后，启动Dispatcher(内部启动了RPC服务)
{
	DefaultDispatcherRunner.java
  	grantLeadership
    	startNewDispatcherLeaderProcess
    		dispatcherLeaderProcess = createNewDispatcherLeaderProcess  			
    		dispatcherLeaderProcess.start

	AbstractDispatcherLeaderProcess.java
  	start
    	startInternal
      	onStart

	JobDispatcherLeaderProcess.java
  	onStart
    	dispatcherGatewayServiceFactory.create

	DefaultDispatcherGatewayServiceFactory.java
  	create(DispatcherId fencingToken, Collection<JobGraph> recoveredJobs, JobGraphWriter jobGraphWriter)
    	{
      	Dispatcher dispatcher = dispatcherFactory.createDispatcher
        	// JobDispatcherFactory.java
          createDispatcher
          	// MiniDispatcher extends Dispatcher
            new MiniDispatcher
            	{
              	// 启动RPC服务，内部会创建Actor，返回代理对象AkkaInvocationHandler
                this.rpcServer = rpcService.startServer(this)
              }
          dispatcher.start         
      }
}

// 3) 启动后做什么？startRecoveredJobs(start JobMasters)  
{
	Dispatcher.java
  	start()
    .. // start() -> onStart() 过程，同上述ResourceManager调用链一致，不再多述
    onStart()
    	startDispatcherServices
      	startRecoveredJobs
        	// 遍历recoveredJobs
          runRecoveredJob(recoveredJob)
          	runJob
            	createJobManagerRunner
              	JobManagerRunner runner = jobManagerRunnerFactory.createJobManagerRunner
                runner.start()

  // implements LeaderContender  
	JobMasterServiceLeadershipRunner.java  
  	start
    	leaderElectionService.start(this)
    	... // 后续 start(LeaderContender) -> leaderContender.grantLeadership过程同上述ResourceManager调用链一致，不再多述
      grantLeadership
      	startJobMasterServiceProcessAsync
        	verifyJobSchedulingStatusAndCreateJobMasterServiceProcess
          	createNewJobMasterServiceProcess
            	jobMasterServiceProcess = jobMasterServiceProcessFactory.create
    
	DefaultJobMasterServiceProcessFactory.java
  	create
    	new DefaultJobMasterServiceProcess
      	this.jobMasterServiceFuture = jobMasterServiceFactory.createJobMasterService
    
	DefaultJobMasterServiceFactory.java
  	createJobMasterService
    	internalCreateJobMasterService
      	JobMaster jobMaster = new JobMaster(...)
        jobMaster.start()

}