flink-on-yarn 基础架构和启动流程

分析flink-on-yarn per-job flink集群整体调用链 flink-1.10

flink-on-yarn
yarn-作业提交流程

• ./flink run -d -m yarn-cluster ./flinkExample.jar
  1.vim bin/flink
  exec $JAVA_RUN $JVM_ARGS "${log_setting[@]}" -classpath manglePathList ``"$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"" org.apache.flink.client.cli.CliFrontend "$@"`
  2.得到入口类
  org.apache.flink.client.cli.CliFrontend[main]

// 3. load the custom command lines
        final List<CustomCommandLine> customCommandLines = loadCustomCommandLines(
            configuration,
            configurationDirectory);
//  Command line interface of the YARN session, with a special initialization here
        //  to prefix all options with y/yarn.
        final String flinkYarnSessionCLI = "org.apache.flink.yarn.cli.FlinkYarnSessionCli";
        try {
            customCommandLines.add(
                loadCustomCommandLine(flinkYarnSessionCLI,
                    configuration,
                    configurationDirectory,
                    "y",
                    "yarn"));
        } catch (NoClassDefFoundError | Exception e) {
            LOG.warn("Could not load CLI class {}.", flinkYarnSessionCLI, e);
        }

parseParameters()->run()->executeProgram()-> PackagedProgram.java program.invokeInteractiveModeForExecution()->
3.调用到主类，用户代码主类
callMainMethod(mainClass, args)

• 用户代码
  flink-stream-application
env.execute()

• flink-client 端代码流程
  flink-streaming-java StreamExecutionEnvironment.java
  execute() streamGraph
  flink-client AbstractJobClusterExecutor.java
  execute()
  1.streamGraph -> JobGraph
  2.deployJobCluster

• flink-yarn
  FlinkYarnSessionCli[main]->run->yarnClusterDescriptor.deploySessionCluster(clusterSpecification)->YarnClusterDescriptor.java.deploySessionCluster->deployInternal

    return deployInternal(
                    clusterSpecification,
                    "Flink session cluster",
                    getYarnSessionClusterEntrypoint(),
                    null,
                    false);

注意这里有获取ClusterEntrypoint的方法，得到启动AM container 之后会调用该类的main方法，启动集群
YarnSessionClusterEntrypoint.java[main] ClusterEntrypoint.runClusterEntrypoint(yarnSessionClusterEntrypoint);->clusterEntrypoint.startCluster();-> ClusterEntrypoint.java runCluster(configuration);
1.检查环境check if required Hadoop environment，HADOOP_CONF_DIR||YARN_CONF_DIR
2.startAppMaster
This method will block until the ApplicationMaster/JobManager have been deployed on YARN
返回-集群客户端ApplicationReport

startAppMaster(
            Configuration configuration,
            String applicationName,
            String yarnClusterEntrypoint,
            JobGraph jobGraph,
            YarnClient yarnClient,
            YarnClientApplication yarnApplication,
            ClusterSpecification clusterSpecification)

yarnClient.submitApplication(appContext);

通过 yarn-client向yarn-RM提交Submitting application master 要求启动AppMaster（AppMaster和jobManger在一个进程中两个线程），返回RestClusterClient flink集群客户端给flink-client，到此为止，应用请求已提交到Yarn的ResourceManager上了

• yarn流程，该部分主要是在NodeManager上启动AM container都是Yarn本身行为
  YarnClientImpl.java
  RMAppManager.java submitApplication()
  this.rmContext.getDispatcher().getEventHandler().handle(new RMAppEvent(applicationId, RMAppEventType.START));
  主要是创建RMAppEvent，向RM申请container

• flink-on-yarn Per-job 集群启动流程分析
  AM container加载运行的入口是 YarnJobClusterEntryPoint.java 中的main()方法
  jobMaster启动流程
  The executable entry point for the Yarn Application Master Process for a single Flink job.
  ClusterEntrypoint.runClusterEntrypoint(yarnJobClusterEntrypoint)->
  ClusterEntrypoint.java-> runClusterEntrypoint()->clusterEntrypoint.startCluster()->runCluster()->

clusterComponent = dispatcherResourceManagerComponentFactory.create(
                configuration,
                ioExecutor,
                commonRpcService,
                haServices,
                blobServer,
                heartbeatServices,
                metricRegistry,
                archivedExecutionGraphStore,
                new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()),
                this);
DefaultDispatcherResourceManagerComponentFactory.java
public DispatcherResourceManagerComponent create(
            Configuration configuration,
            Executor ioExecutor,
            RpcService rpcService,
            HighAvailabilityServices highAvailabilityServices,
            BlobServer blobServer,
            HeartbeatServices heartbeatServices,
            MetricRegistry metricRegistry,
            ArchivedExecutionGraphStore archivedExecutionGraphStore,
            MetricQueryServiceRetriever metricQueryServiceRetriever,
            FatalErrorHandler fatalErrorHandler){
       //资源管理器
    resourceManager = resourceManagerFactory.createResourceManager(
                configuration,
                ResourceID.generate(),
                rpcService,
                highAvailabilityServices,
                heartbeatServices,
                fatalErrorHandler,
                new ClusterInformation(hostname, blobServer.getPort()),
                webMonitorEndpoint.getRestBaseUrl(),
                resourceManagerMetricGroup);

            //注意这里就是启动Dispatcher rpc服务的
            log.debug("Starting Dispatcher REST endpoint.");
            webMonitorEndpoint.start();
            log.debug("Starting Dispatcher.");
            dispatcherRunner = dispatcherRunnerFactory.createDispatcherRunner(
                highAvailabilityServices.getDispatcherLeaderElectionService(),
                fatalErrorHandler,
                new HaServicesJobGraphStoreFactory(highAvailabilityServices),
                ioExecutor,
                rpcService,
                partialDispatcherServices);

            log.debug("Starting ResourceManager.");
            resourceManager.start();
            
            resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);
            //启动 dispathcer 选举服务，同时也启动 dispathcer  Zookeeper开源客户端Curator实现leader选举逻辑 flink选择的方式leaderLatch选举
            dispatcherLeaderRetrievalService.start(dispatcherGatewayRetriever);->ZooKeeperLeaderRetrievalService.java start()
            
    return new DispatcherResourceManagerComponent(
                dispatcherRunner,
                resourceManager,
                dispatcherLeaderRetrievalService,
                resourceManagerRetrievalService,
                webMonitorEndpoint);
}

在同一进程中启动Dispatcher，ResourceManager和WebMonitorEndpoint组件服务
方法里面启动了 resourcemanager 和 dispatcher ，而Dispatcher类就是客户端提交Job的入口,参见 Dispatcher 注释,具体来说是其中的submitJob方法
JobMaster 是负责单个 JobGraph 的执行的,JobManager是老的runtime框架，JobMaster是社区 flip-6引入的新的runtime框架。目前起作用的应该是JobMaster

• 分析jobMaster中的ResourceManager组件
  public class YarnResourceManager extends ActiveResourceManager<YarnWorkerNode> implements AMRMClientAsync.CallbackHandler, NMClientAsync.CallbackHandler
  该组件实现了yarn 的 AMRMClientAsync.CallbackHandler接口，在Container分配完之后，回调
  onContainersAllocated方法->requiredContainers.forEach(this::startTaskExecutorInContainer)启动taskManagerContainer->

ContainerLaunchContext taskExecutorLaunchContext = Utils.createTaskExecutorContext(
            flinkConfig,
            yarnConfig,
            env,
            taskManagerParameters,
            taskManagerDynamicProperties,
            currDir,
//taskmanager container 运行类
            YarnTaskExecutorRunner.class,
            log);

• 分析 dispathcer 组件
  Flink中JobMaster、ResourceManager、Dispatcher、WebMonitorEndpoint提供了基于zookeeper高可用
  接上代码 Starting Dispatcher 分析 createDispatcherRunner->DefaultDispatcherRunnerFactory.java createDispatcherRunner ->DefaultDispatcherRunner.java create() ->DispatcherRunnerLeaderElectionLifecycleManager.createFor(dispatcherRunner, leaderElectionService)->return new DispatcherRunnerLeaderElectionLifecycleManager<>(dispatcherRunner, leaderElectionService) ->DispatcherRunnerLeaderElectionLifecycleManager.leaderElectionService.start(dispatcherRunner) -> ZooKeeperLeaderElectionService.java start() leaderLatch.addListener(this) leaderLatch.start()启动选举
  一旦选主成功，会调用 ZooKeeperLeaderElectionService isLeader()->leaderContender.grantLeadership(issuedLeaderSessionID)-> JobManagerRunnerImpl extends JobManagerRunner ->grantLeadership() verifyJobSchedulingStatusAndStartJobManager
  到这里找到jobMaster的启动方法，grantLeadership 实现类中有 JobManagerRunner, ResourceManager 、DefaultDispatcherRunner、WebMonitorEndpoint 里面都是jobMaster相关主件的启动启动方法

private CompletableFuture<Void> verifyJobSchedulingStatusAndStartJobManager(UUID leaderSessionId) {
        final CompletableFuture<JobSchedulingStatus> jobSchedulingStatusFuture = getJobSchedulingStatus();

        return jobSchedulingStatusFuture.thenCompose(
            jobSchedulingStatus -> {
                if (jobSchedulingStatus == JobSchedulingStatus.DONE) {
                    return jobAlreadyDone();
                } else {
                    return startJobMaster(leaderSessionId);
                }
            });
    }

这里使用了jdk1.8 CompletableFuture.thenthenCompose特性，从zk获取job.graph文件的hdfs路径然后读取为JobGraph->JobManagerRunnerImpl.java startJobMaster-> return startJobMaster(leaderSessionId)
->startFuture = jobMasterService.start(new JobMasterId(leaderSessionId));
->JobMaster.java start();
->startJobExecution(newJobMasterId)
->startJobMasterServices
->resetAndStartScheduler
->由jobGraph创建ExecutionGraph
final SchedulerNG newScheduler = createScheduler(newJobManagerJobMetricGroup)
->JobMaster.java createScheduler->SchedulerNGFactory->createInstance->LegacySchedulerFactory.java createInstance->LegacyScheduler->SchedulerBase.java createAndRestoreExecutionGraph
createExecutionGraph(currentJobManagerJobMetricGroup, shuffleMaster, partitionTracker)，成功创建ExecutionGraph
-jobMaster.java ->schedulerAssignedFuture.thenRun(this::startScheduling);
schedulerNG.startScheduling()->SchedulerBase.java startScheduling()->startSchedulingInternal()
->LegacyScheduler.java startSchedulingInternal()->ExecutionGraph.java executionGraph.scheduleForExecution();
-> final CompletableFuture<Void> newSchedulingFuture = SchedulingUtils.schedule(
scheduleMode,
getAllExecutionVertices(),
this);
scheduleLazy()->ExecutionVertex.scheduleForExecution()->Execution.scheduleForExecution()->this::deploy()-->taskManagerGateway.submitTask(deployment, rpcTimeout), executor)
->RpcTaskManagerGateway.submitTask(TaskDeploymentDescriptor tdd, Time timeout)
->TaskExecutor.submitTask()->task.startTaskThread();->Task.java impl Thread executingThread.start()->run()->doRun()->AbstractInvokable invokable.invoke()
run():The core work method that bootstraps the task and executes its code.
AbstractInvokable invokable.invoke():he TaskManager invokes the {@link #invoke()} method when executing a task.
->invoke->runMailboxLoop():便开始处理Source端消费的数据，并流入下游算子处理，也就是执行ExecutionVertex生成者逻辑->IntermediateResult (中间结果集合)ExecutionEdge(消费者)

分析 runMailboxLoop

OneInputStreamTask.java extends StreamTask

public OneInputStreamTask(
            Environment env,
            @Nullable TimerService timeProvider) {
        super(env, timeProvider);
    }

protected StreamTask(
            Environment environment,
            @Nullable TimerService timerService,
            Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
            StreamTaskActionExecutor.SynchronizedStreamTaskActionExecutor actionExecutor,
            TaskMailbox mailbox) {

        super(environment);
...
        this.mailboxProcessor = new MailboxProcessor(this::processInput, mailbox, actionExecutor);
...
    }
// This method implements the default action of the task (e.g. processing one event from the input)
protected void processInput(MailboxDefaultAction.Controller controller){
      InputStatus status = inputProcessor.processInput();
}
StreamOneInputProcessor#processInput
@Override
    public InputStatus processInput() throws Exception {
                注意这里
        InputStatus status = input.emitNext(output);

        if (status == InputStatus.END_OF_INPUT) {
            synchronized (lock) {
                operatorChain.endHeadOperatorInput(1);
            }
        }

        return status;
    }
StreamTaskNetworkInput.java ->emitNext(DataOutput<T> output)
->processElement(deserializationDelegate.getInstance(), output)
->private void processElement(StreamElement recordOrMark, DataOutput<T> output) throws Exception {
        if (recordOrMark.isRecord()){
            output.emitRecord(recordOrMark.asRecord());
        } else if (recordOrMark.isWatermark()) {
            statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), lastChannel);
        } else if (recordOrMark.isLatencyMarker()) {
            output.emitLatencyMarker(recordOrMark.asLatencyMarker());
        } else if (recordOrMark.isStreamStatus()) {
            statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), lastChannel);
        } else {
            throw new UnsupportedOperationException("Unknown type of StreamElement");
        }
    }
->OneInputStreamTask.java emitRecord(StreamRecord<IN> record) 
->public void emitRecord(StreamRecord<IN> record) throws Exception {
            synchronized (lock) {
                numRecordsIn.inc();
                operator.setKeyContextElement1(record);
                operator.processElement(record);
            }
        }

StreamMap.java
processElement->StreamMap.java ->processElement(StreamRecord<IN> element)
->output.collect(element.replace(userFunction.map(element.getValue())))
processElement的具体实现都是flink的各个算子
到这里终于找到用户定义函数执行的地方，到此，整个作业执行都已经理清楚

• 再分析：runMailboxLoop()->runMailboxLoop()
  注意这里执行默认动作的地方-也是真正执行用户udf的地方
  mailboxDefaultAction.runDefaultAction(defaultActionContext)调起过程
  this.mailboxProcessor = new MailboxProcessor(this::processInput, mailbox, actionExecutor)
  StreamTask.java#runMailboxLoop

public void runMailboxLoop() throws Exception {

        final TaskMailbox localMailbox = mailbox;

        Preconditions.checkState(
            localMailbox.isMailboxThread(),
            "Method must be executed by declared mailbox thread!");

        assert localMailbox.getState() == TaskMailbox.State.OPEN : "Mailbox must be opened!";

        final MailboxController defaultActionContext = new MailboxController(this);

        while (processMail(localMailbox)) {
//注意这里执行默认动作的地方-也是真正执行用户udf的地方
            mailboxDefaultAction.runDefaultAction(defaultActionContext); // lock is acquired inside default action as needed
        }

总结：

1.先启动AMMaster container,然后启动内部resourceManager(负责和yarn-RS请求资源，并且taskManagercontainer会在远程启动)，再启动WebMonitorEndpoint，最后启动Dispatcher组件
2.flink cluster on yarn 的方式分析-Dispatcher启动之后，会启动JobMaster，并且会读取job.graph文件解析为ExecutionGraph，然后JobMaster allocateResourcesForExecution，再远程taskManager进行作业的部署
3.还有一种LeaderRetrievalHandler netty channelRead0 ，client通过提交jar包的方式，启动jobMaster，JobSubmitHandler.java handleRequest，上传jar，配置文件等，然后submit jobgraph，Dispatcher.java submitJob,最后实现都是JobManagerRunnerImpl.java
参考
https://zhuanlan.zhihu.com/p/83141161
https://cloud.tencent.com/developer/article/1586184
yarn 架构
https://matt33.com/2018/09/01/yarn-architecture-learn/
flink zookeeper curator
https://blog.csdn.net/u013516966/article/details/103867207
https://blog.csdn.net/weixin_41608066/article/details/105566489