Flink源码解析之：Flink On Yarn模式任务提交部署过程解析

Flink源码解析之：Flink On Yarn模式任务提交部署过程解析

一、Flink on Yarn部署模式概述

Apache Hadoop YARN 在许多数据处理框架中都很流行。 Flink 服务提交给 YARN 的 ResourceManager，后者会在 YARN NodeManagers 管理的机器上生成容器。 Flink 将其 JobManager 和 TaskManager 实例部署到这些容器中。

Flink 可根据在 JobManager 上运行的作业所需的处理插槽数量，动态分配和取消分配 TaskManager 资源。

Flink on Yarn的部署模式包括三种方式，Application Mode、Per-Job Mode、Session Mode。对于生成环境来说，更推荐使用Application Mode或Per-Job Mode，因为这两种模式能够提供更好的应用隔离性。

  1. Application Mode
    Application Mode模式将在 YARN 上启动一个 Flink 集群，应用程序 jar 的 main() 方法将在 YARN 的 JobManager 上执行。 应用程序一旦完成，群集就会关闭。 该种方式相比Per-Job模式来说，将应用main()方法的执行，StreamGraph、JobGraph的生成放在了Flink集群侧来实现。
  2. Per-Job Mode
    Per-job 模式将在 YARN 上启动一个 Flink 集群，在客户端生成StreamGraph、JobGraph，并上传依赖项。最后将 JobGraph 提交给 YARN 上的 JobManager。 如果通过—detached参数配置了分离模式，则客户端将在提交被接受后立即停止。
  3. Session Mode
    Session部署模式会在YARN上部署一个长期运行的Flink集群会话，该会话可以接受并执行多个Flink作业。
    Session部署模式包含两种操作模式：
    • attach mode（default）：执行yarn-session.sh文件在Yarn上启动Flink集群，启动后客户端会一致运行，来追踪/监听集群状态。一旦集群异常，客户端会获取异常信息并展示。如果客户端异常终止了，则会发送signal到Flink集群，此时Flink集群同样也会终止。
    • detach mode ：使用-d or --detached参数设置。在这种模式下，当执行yarn-session.sh文件在Yarn上启动Flink集群后，客户端会直接返回。要停止 Flink 群集，需要再次调用客户端或 YARN 工具。

三种提交模式的对比：
由bin/flink.sh脚本可知，客户端提交过程统一由org.apache.flink.client.cli.CliFronted入口类触发。Per-Job模式和Session模式下Flink应用main方法都会在客户端执行。客户端解析生成JobGraph后会将依赖项和JobGraph序列化后的二进制数据一起发往集群上。当客户端机器上有大量作业提交时，需要大量的网络带宽下载依赖项并将二进制文件发送到集群，会造成客户端消耗大量的资源。尤其在大量用户共享客户端时，问题更加突出。为解决该问题，社区提出了Application模式将Flink应用main方法触发过程后置到了JobManager生成过程中，以此将带宽压力分散到集群各个节点上。

鉴于Application部署模式的优势，本文会以Application部署模式的源码来进行解析，探究Flink以Application模式提交任务到Yarn集群中所经过的大致流程，为我们理解Flink On Yarn的部署有一个更深入和清晰的认识。

二、Flink Application部署模式源码解析

（一）CliFronted入口类

本节以Application部署模式为例，介绍Flink On Yarn的客户端提交源码流程。正如上文说的，由bin/flink.sh脚本可知，客户端提交过程统一由org.apache.flink.client.cli.CliFronted入口类触发，为此，我们首先进入到该方法的源码中，来观察下该入口类的实现逻辑：

/** Submits the job based on the arguments. */
public static void main(final String[] args) {EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);// 1. find the configuration directory// 用来获取配置目录，该目录通常包含Flink的配置文件，如flink-conf.yaml。final String configurationDirectory = getConfigurationDirectoryFromEnv();// 2. load the global configuration// 加载flink的全局配置final Configuration configuration =GlobalConfiguration.loadConfiguration(configurationDirectory);// 3. load the custom command lines// 加载自定义的命令行配置final List<CustomCommandLine> customCommandLines =loadCustomCommandLines(configuration, configurationDirectory);int retCode = 31;try {// 实例化了CliFronted对象，CliFronted是Flink为CLI客户端提供的API，它提供了一系列的操作，例如作业的提交，取消，以及打印job的状态等。final CliFrontend cli = new CliFrontend(configuration, customCommandLines);SecurityUtils.install(new SecurityConfiguration(cli.configuration));// 启动Flink作业的入口，parseAndRun方法会解析命令行参数，并启动Flink作业。retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));} catch (Throwable t) {final Throwable strippedThrowable =ExceptionUtils.stripException(t, UndeclaredThrowableException.class);LOG.error("Fatal error while running command line interface.", strippedThrowable);strippedThrowable.printStackTrace();} finally {System.exit(retCode);}
}

上述代码是CliFronted的入口main方法，该方法首先根据Flink的配置路径加载全局配置，比如flink-conf.xml配置文件，接着加载自定义命令行配置，并实例化了CliFronted对象。CliFronted是Flink为CLI客户端提供的API，它提供了一系列的操作，例如作业的提交，取消，以及打印job的状态等。最后，调用cli.parseAndRun(args)方法，该方法会解析命令行参数，并启动Flink作业。

在parseAndRun方法中，会根据传入参数的第一个参数值来决定Flink集群的部署模式：

  • run-application：则会进入到CliFronted类的runApplication方法中，执行Application部署流程。
  • run：则会进入到CliFronted类的run方法中，在客户端执行作业的main方法（利用反射来执行）

这也是为什么我在使用命令行以Application模式部署Flink集群时，命令的开始要用以下形式：

/bin/flink run-application -t yarn-application...

（二）runApplication

接下来，我们继续进入到runApplicaiton方法来看看它的实现逻辑：

protected void runApplication(String[] args) throws Exception {LOG.info("Running 'run-application' command.");// 解析传入的命令行参数final Options commandOptions = CliFrontendParser.getRunCommandOptions();final CommandLine commandLine = getCommandLine(commandOptions, args, true);// 如果命令行参数中包含帮助选项（-h/--help），则调用下述方法打印帮助信息并返回if (commandLine.hasOption(HELP_OPTION.getOpt())) {CliFrontendParser.printHelpForRunApplication(customCommandLines);return;}// 验证并获取激活的自定义命令行， CustonCommandLine是Flink用来处理不同部署模式的工具（例如Yarn,Standlone等），以便针对不同模式解析对应的特定设置和参数final CustomCommandLine activeCommandLine =validateAndGetActiveCommandLine(checkNotNull(commandLine));// 初始化ApplicationClusterDeployer实例， 这是Flink用来启动Application的工具final ApplicationDeployer deployer =new ApplicationClusterDeployer(clusterClientServiceLoader);final ProgramOptions programOptions;final Configuration effectiveConfiguration;// No need to set a jarFile path for Pyflink job.if (ProgramOptionsUtils.isPythonEntryPoint(commandLine)) {programOptions = ProgramOptionsUtils.createPythonProgramOptions(commandLine);effectiveConfiguration =getEffectiveConfiguration(activeCommandLine,commandLine,programOptions,Collections.emptyList());} else {programOptions = new ProgramOptions(commandLine);programOptions.validate();final URI uri = PackagedProgramUtils.resolveURI(programOptions.getJarFilePath());effectiveConfiguration =getEffectiveConfiguration(activeCommandLine,commandLine,programOptions,Collections.singletonList(uri.toString()));}// 根据programOptions获取程序参数和入口类名来创建ApplicationConfiguration实例final ApplicationConfiguration applicationConfiguration =new ApplicationConfiguration(programOptions.getProgramArgs(), programOptions.getEntryPointClassName());// 最后调用deployer.run()来运行应用。这一步通常包括联系Flink集群，提交应用程序并安排其在集群中执行。deployer.run(effectiveConfiguration, applicationConfiguration);
}

上述代码的实现流程与原理如下所示：

  • 解析命令行参数：首先，调用getCommandLine函数解析传入的命令行参数args。

  • 处理帮助选项：如果命令行参数中包含帮助选项(-h/–help)，则调用CliFrontendParser.printHelpForRunApplication打印帮助信息并返回。

  • 获取激活的CustomCommandLine：通过validateAndGetActiveCommandLine函数获取激活的自定义命令行（CustomCommandLine）。CustomCommandLine是Flink用来处理不同部署模式的工具（例如Yarn，Standalone等），以便于针对不同模式解析对应的特定设置和参数。

  • 部署器配置：初始化ApplicationClusterDeployer实例，这是Flink用来启动Application的工具。

  • 提取程序选项和计算有效配置：区分Python作业和其他作业，生成对应的ProgramOptions并验证其有效性。此外，根据激活的命令行、解析得到的命令行参数和程序选项计算出有效的配置（effectiveConfiguration）。

  • 构造应用配置：使用从ProgramOptions中获取的程序参数和入口点类名创建ApplicationConfiguration实例。

  • 运行应用：最后，调用deployer.run()来运行应用。这一步通常包括联系Flink集群，提交应用程序并安排其在集群中执行。

ProgramOptions.entryPointClass的成员值是flink命令行 -c 选项指定的Flink应用入口类com.xxx.xxx.FlinkApplicationDemo，后续会以反射的形式触发main()方法的执行。

（三）ClusterDescriptor.deployApplicationCluster

上面代码中deployer.run(...)方法负责加载Yarn Application模式客户端信息等。

首先代码会根据configuration配置信息来获取ClusterClientFactory对象，获取的逻辑过程是根据configuration配置中的execution.target参数来决定的。

当执行命令行bin/flink run时， execution.target参数对应的枚举值可以如下：

  • remote
  • local
  • yarn-per-job
  • yarn-session
  • kubernetes-session
    当执行命令行bin/flink run-application时，execution.target参数对应的枚举值可以如下：
  • yarn-application
  • kubernetes-application

当execution.target参数为yarn-application时，Flink便会生成相应的YarnClusterClientFactory客户端工厂类，然后调用该工厂类的createClusterDescriptor方法，该方法中会新建YarnClient实例，YarnClient实例负责在客户端提交Flink应用程序，并最终生成ClusterDescriptor实例，该实例包含用于在Yarn上部署Flink集群的部署信息Descriptor。

@Override
public YarnClusterDescriptor createClusterDescriptor(Configuration configuration) {checkNotNull(configuration);final String configurationDirectory = configuration.get(DeploymentOptionsInternal.CONF_DIR);YarnLogConfigUtil.setLogConfigFileInConfig(configuration, configurationDirectory);return getClusterDescriptor(configuration);
}private YarnClusterDescriptor getClusterDescriptor(Configuration configuration) {final YarnClient yarnClient = YarnClient.createYarnClient();final YarnConfiguration yarnConfiguration =Utils.getYarnAndHadoopConfiguration(configuration);yarnClient.init(yarnConfiguration);yarnClient.start();return new YarnClusterDescriptor(configuration,yarnConfiguration,yarnClient,YarnClientYarnClusterInformationRetriever.create(yarnClient),false);
}

有了该实例后，会调用deployApplicationCluster方法来部署Application模式的Flink集群。集群将在提交应用程序时创建，并在应用程序终止时拆除。此外，应用程序用户代码的{@code main()}将在集群上执行，而不是在客户端上执行。

YarnClusterDescriptor.deployApplicationCluster(…)方法调用过程如下：
(1)、YarnClusterDescriptor.deployApplicationCluster(…);进行一些配置和检查，并调用deployInternal(…)方法。
(2)、YarnClusterDescriptor.deployInternal(…);

其中，最重要的方法是deployInternal方法

（四）YarnClusterDescriptor.deployInternal

在该方法中，首先会判断Hadoop集群是否启用了Kerberos安全认证，如果开启了，则Flink会首先确认当前用户是否拥有有效的kerberos凭证。如果无效，则会抛出异常，部署作业失败。

紧接着，进行资源检查和部署模式判断。

在validateClusterResources方法中，会根据配置的JobManager和TaskManager的资源大小与集群资源进行比对。

  1. 如果JobManager的配置内存大小 < Yarn配置的最小调度分配内存(yarn.scheduler.minimum-allocation-mb参数，默认1024MB)，则JobManager的内存大小会设置为该配置值。
  2. 如果JobManager大小 > YARN 集群能够提供的单个容器的最大资源，则抛出异常：The cluster does not have the requested resources for the JobManager available!
  3. 如果TaskManager大小 > YARN 集群能够提供的单个容器的最大资源，则抛出异常：The cluster does not have the requested resources for the TaskManagers available!
  4. 如果TaskManager大小 > 当前YARN集群剩余资源单个任务容器分配的最大资源容量，则会打印告警日志：The requested amount of memory for the TaskManagers is more than the largest possible YARN container: freeClusterResources.containerLimit
  5. 如果JobManager大小 > 当前YARN集群剩余资源单个任务容器分配的最大资源容量，则会打印告警日志：The requested amount of memory for the JobManager is more than the largest possible YARN container: freeClusterResources.containerLimit

经过资源检查后，会将最后确定的JobManager和TaskManager资源保存在ClusterSpecification对象中。

在部署模式决定中，Flink 提供了两种部署模式：Detach模式和Non-Detach模式。如果是 Detach模式，Flink 作业提交到YARN后，客户端可以直接退出，而作业将继续在YARN集群上运行。而在 Non-Detach模式下，客户端将持续等待作业执行完成。

然后就到了一个非常重要的方法中：startAppMaster。

会根据上面决定的ClusterSpecification资源实例，启动用于管理Flink作业的Application Master。

startAppMaster方法比较长。

这段代码主要是用于启动Flink在YARN集群上的Application Master的过程，代码中包含了几个主要部分：

  1. 首先，核心的首步骤是初始化文件系统并获取对应的 FileSystem 实例。代码检查了文件系统的类型，如果是本地文件系统类型（file://开头），会抛出警告，因为Flink在YARN上运行需要分布式文件系统来存储文件。
  2. 然后，获取了用于提交应用程序的 ApplicationSubmissionContext，并将 Flink 应用所需的各种文件如jar包、配置文件等上传到HDFS，并将这些文件的HDFS路径作为本地资源 （LocalResources）添加到ApplicationSubmissionContext里。
  3. 在文件上传阶段，包括了一系列复杂的步骤，首先是将 flink 配置、job graph、用户 jar、依赖库等上传到HDFS，并将这些文件的路径添加到应用的classpath；其次，如果设置了 security options（例如，Kerberos认证信息），会将相关文件也上传到HDFS；并且，对配置了Kerberos认证的 flink 应用，会从 YARN 获取 HDFS delegation tokens。
  4. 在收集完上述一系列依赖文件后，final ContainerLaunchContext amContainer = setupApplicationMasterContainer(yarnClusterEntrypoint, hasKrb5, processSpec) 负责设置启动ApplicationMaster的命令操作。
  5. 设置ApplicationMaster的环境变量，诸如_FLINK_CLASSPATH、_FLINK_DIST_JAR(Flink jar resource location (in HDFS))、KRB5_PATH、_YARN_SITE_XML_PATH等环境变量。最后调用amContainer.setEnvironment(appMasterEnv);方法进行设置。
  6. 接着，会将上述配置好的amContainer实例放入ApplicationSubmissionContext对象中，以及ApplicationName和所需的资源大小，最终交给交给YarnClient去提交，并随后通过周期性地获取应用状态，来等待应用处于RUNNING或FINISHED状态，完成应用的提交过程。
  7. 如果在这一系列操作中有任何异常或错误发生，会触发失败保护钩子 DeploymentFailureHook，进行必要的清理工作。

上面这段代码体现了 Flink on YARN 的工作原理，Flink 通过 YARN Client 提交应用，启动 Application Master 来进行资源申请和任务调度，这是典型的 YARN 应用程序模型。各种文件（包括 flink 本身、用户 jar、配置文件等）都被上传到HDFS，然后再从HDFS分发到运行任务的 YARN 容器中，这样做是为了实现文件的分布式共享，并且利用了 YARN 的 LocalResource 机制来进行文件的分发。

对于第四点中的setupApplicationMasterContainer方法，该方法构造了ApplicationMaster的命令行启动命令，如下所示：

ContainerLaunchContext setupApplicationMasterContainer(String yarnClusterEntrypoint, boolean hasKrb5, JobManagerProcessSpec processSpec) {// ------------------ Prepare Application Master Container  ------------------------------// respect custom JVM options in the YAML fileString javaOpts = flinkConfiguration.getString(CoreOptions.FLINK_JVM_OPTIONS);if (flinkConfiguration.getString(CoreOptions.FLINK_JM_JVM_OPTIONS).length() > 0) {javaOpts += " " + flinkConfiguration.getString(CoreOptions.FLINK_JM_JVM_OPTIONS);}// krb5.conf file will be available as local resource in JM/TM containerif (hasKrb5) {javaOpts += " -Djava.security.krb5.conf=krb5.conf";}// Set up the container launch context for the application masterContainerLaunchContext amContainer = Records.newRecord(ContainerLaunchContext.class);final Map<String, String> startCommandValues = new HashMap<>();startCommandValues.put("java", "$JAVA_HOME/bin/java");String jvmHeapMem =JobManagerProcessUtils.generateJvmParametersStr(processSpec, flinkConfiguration);startCommandValues.put("jvmmem", jvmHeapMem);startCommandValues.put("jvmopts", javaOpts);startCommandValues.put("logging", YarnLogConfigUtil.getLoggingYarnCommand(flinkConfiguration));startCommandValues.put("class", yarnClusterEntrypoint);startCommandValues.put("redirects","1> "+ ApplicationConstants.LOG_DIR_EXPANSION_VAR+ "/jobmanager.out "+ "2> "+ ApplicationConstants.LOG_DIR_EXPANSION_VAR+ "/jobmanager.err");String dynamicParameterListStr =JobManagerProcessUtils.generateDynamicConfigsStr(processSpec);startCommandValues.put("args", dynamicParameterListStr);final String commandTemplate =flinkConfiguration.getString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE,ConfigConstants.DEFAULT_YARN_CONTAINER_START_COMMAND_TEMPLATE);final String amCommand =BootstrapTools.getStartCommand(commandTemplate, startCommandValues);amContainer.setCommands(Collections.singletonList(amCommand));LOG.debug("Application Master start command: " + amCommand);return amContainer;
}

启动命令的参数包括以下部分：

  • “java”：Java二进制文件的路径。一般来说，在YARN容器中，Java的路径会被设置为$JAVA_HOME/bin/java。
  • “jvmmem”：JVM参数，主要是内存参数，比如最大堆内存、最小堆内存等。这些参数会基于Flink配置以及JobManager的内存配置来生成。
  • “jvmopts”：JVM选项。这些选项来自Flink配置文件中设置的JVM选项，以及若存在Kerberos krb5.conf文件，还会添加-Djava.security.krb5.conf=krb5.conf。
  • “logging”：日志配置项，用于配置Flink的日志选项。
  • “class”：启动类，即YARN集群入口点类名（yarnClusterEntrypoint）。
  • “redirects”：输出重定向的参数，将stdout（输出流）和stderr（错误流）重定向到日志文件中。
  • “args”：传递给启动类的参数，主要是JobManager的动态配置参数。

⠀这些参数最后会填入一个启动命令模板（通常为"%java% %jvmmem% %jvmopts% %logging% %class% %args% %redirects%"），来生成实际启动Flink应用的命令。

启动后的ApplicationMaster，在YARN集群上起着以下的关键作用：

  • 作为应用程序的主控制器，管理和监视应用程序的执行。
  • 负责请求YARN ResourceManager分配所需的资源（例如容器）。
  • 启动和监视任务执行器(TaskExecutor)，它们在分配的容器中运行。
  • 与Flink的client（例如命令行界面或Web界面）以及ResourceManager进行交互，提供应用程序的状态和进度信息。
  • 在应用程序出现异常或失败时，它可以选择重新请求资源并重启失败的任务，提供了一定程度的错误恢复能力。

⠀因此，Application Master是Flink在YARN上运行的关键组件，它负责管理Flink应用程序的生命周期和资源。

（五）YarnApplicationClusterEntryPoint

在上面的setupApplicationMasterContainer方法中，我们说该方法构建了ApplicationMaster的启动命令。从该命令行中可以看到，命令行的启动入口类为yarnClusterEntrypoint参数，对于Yarn Application部署模式来说，参数对应的入口类即为YarnApplicationClusterEntryPoint。在第四部分的分析中，当通过yarnClient将ApplicationMaster提交到Yarn集群后，便会申请Container来执行ApplicationMaster，执行该入口类。
为此，接下来我们来分析一下，YarnApplicationClusterEntryPoint入口类的执行逻辑。

public static void main(final String[] args) {// startup checks and loggingEnvironmentInformation.logEnvironmentInfo(LOG, YarnApplicationClusterEntryPoint.class.getSimpleName(), args);SignalHandler.register(LOG);JvmShutdownSafeguard.installAsShutdownHook(LOG);Map<String, String> env = System.getenv();// 获取工作路径final String workingDirectory = env.get(ApplicationConstants.Environment.PWD.key());Preconditions.checkArgument(workingDirectory != null,"Working directory variable (%s) not set",ApplicationConstants.Environment.PWD.key());try {YarnEntrypointUtils.logYarnEnvironmentInformation(env, LOG);} catch (IOException e) {LOG.warn("Could not log YARN environment information.", e);}final Configuration dynamicParameters =ClusterEntrypointUtils.parseParametersOrExit(args,new DynamicParametersConfigurationParserFactory(),YarnApplicationClusterEntryPoint.class);final Configuration configuration =YarnEntrypointUtils.loadConfiguration(workingDirectory, dynamicParameters, env);PackagedProgram program = null;try {// 获取用户应用程序jar，程序参数、入口类名等信息，封装为PackagedProgram实例program = getPackagedProgram(configuration);} catch (Exception e) {LOG.error("Could not create application program.", e);System.exit(1);}try {configureExecution(configuration, program);} catch (Exception e) {LOG.error("Could not apply application configuration.", e);System.exit(1);}YarnApplicationClusterEntryPoint yarnApplicationClusterEntrypoint =new YarnApplicationClusterEntryPoint(configuration, program);// 	执行Application Cluster
ClusterEntrypoint.runClusterEntrypoint(yarnApplicationClusterEntrypoint);
}

上面这段代码中，使用getPackagedProgram(configuration)方法获取用户应用程序jar，程序参数、入口类名等信息，封装为PackagedProgram实例，便于后续调用。

最后，调用runClusterEntrypoint方法，启动执行Application Cluster集群。

ClusterEntrypoint.runClusterEntrypoint(...)方法的调用链路如下：

  • ClusterEntrypoint.runClusterEntrypoint(...)
  • ClusterEntrypoint.startCluster(...)
  • ClusterEntrypoint.runCluster(...)
  • DispatcherResourceManagerComponentFactory.create(…)

在DispatcherResourceManagerComponentFactory.create方法中，启动了一系列服务，比如：

  • LeaderRetrievalService
  • WebMonitorEndpoint
  • ResourceManagerService
  • DispatcherRunner

本流程中主要需要关注的服务是DispatcherRunner，该方法中，会调用dispatcherRunnerFactory.createDispatcherRunner来初始化dispatchRunner实例，dispatcherRunner实例负责dispatcher组件的高可用leader选举操作，同时dispatcher组件负责触发Flink用户应用main(…)方法执行。

在创建DispatchRunner的过程中，包含高可用Leader选举过程，经过一系列的方法链调用，会选举出一个Leader DispatchRunner服务来负责后续的处理流程。

  • DispatcherResourceManagerComponentFactory.createDispatcherRunner
  • DefaultDispatcherRunner.create()
  • DispatcherRunnerLeaderElectionLifecycleManager.createFor()
  • DefaultLeaderElectionService.start()
  • LeaderElectionDriverFactory.createLeaderElectionDriver()
  • new ZooKeeperLeaderElectionDriver
  • LeaderLatch.start()
  • LeaderLatch.internalStart()
  • LeaderLatch.reset()
  • LeaderLatch.setLeadership()
  • ZooKeeperLeaderElectionDriver.isLeader()
  • DefaultLeaderElectionService.onGrantLeadership()
  • DefaultDispatcherRunner.grantLeadership()
  • DefaultDispatcherRunner.startNewDispatcherLeaderProcess()

选举为leader的DefaultDispatcherRunner实例候选者在回调动作过程中会一直调用到上面的grantLeadership(…)方法，并在startNewDispatcherLeaderProcess(…)方法中生成dispatcherLeaderProcess，表示一个Ledaer Dispatcher进程来提供服务，并通过newDispatcherLeaderProcess::start方法来启动执行该服务的后续处理流程。Leader候选者回调动作触发过程会另起篇幅详细讲解，此处先这样理解。

在后续的处理流程中，我们需要关注的点是在何时触发用户应用程序的main方法执行，为此，继续深入以下调用链：

  • AbstractDispatcherLeaderProcess.startInternal()
  • SessionDispatcherLeaderProcess.onStart()
  • SessionDispatcherLeaderProcess.createDispatcherIfRunning()
  • SessionDispatcherLeaderProcess.createDispatcher()
  • ApplicationDispatcherGatewayServiceFactory.create()
  • new ApplicationDispatcherBootstrap(...)

上述调用链中，createDispatcher(…)方法会调用dispatcherGatewayServiceFactory.create(…)方法，dispatcherGatewayServiceFactory实际类型是ApplicationDispatcherGatewayServiceFactory。在dispatcherGatewayServiceFactory.create(…)方法中新建ApplicationDispatcherBootstrap实例。

在ApplicationDispatcherBootstrap实例中，继续通过以下方法调用链fixJobIdAndRunApplicationAsync(…) -> runApplicationAsync(…) -> runApplicationEntryPoint(…) -> ClientUtils.executeProgram(…) -> program.invokeInteractiveModeForExecution() -> callMainMethod(mainClass, args) -> mainMethod.invoke(null, (Object) args)触发Flink应用main(…)方法的执行。

  • ApplicationDispatcherBootstrap.fixJobIdAndRunApplicationAsync()
  • ApplicationDispatcherBootstrap.runApplicationAsync()
  • ApplicationDispatcherBootstrap.runApplicationEntryPoint()
  • ClientUtils.executeProgram()
  • PackagedProgram.invokeInteractiveModeForExecution()
  • PackagedProgram.callMainMethod()
  • mainMethod.invoke(null, (Object) args);

最终，在ApplicationDispatcherBootstrap类的实现中，我们找到了用户应用程序的main方法执行入口。

三、回顾与总结

回顾一下上面的整体流程，首先，我们通过ApplicationMaster的启动命令，找到AM组建执行的入口类为YarnApplicationClusterEntryPoint，接着，在启动集群时，我们发现Flink会初始化一些诸如LeaderRetrievalService、WebMonitorEndpoint、ResourceManagerService、DispatcherRunner的服务，这些服务分别发挥不同的用途，与Yarn和Flink集群进行交互。在本次分析过程中，我们着重探究了DispatcherRunner服务的创建流程。

首先，会执行高可用的选举流程，最终选举出一个Leader DispatcherRunner来执行服务。选举完成后，该Leader DispatchRunner会调用ClientUtils.executeProgram方法，从封装好的PackagedProgram实例中，获取用户应用程序的入口类mainClass以及程序入参，并最终利用反射触发mainClass的main方法的执行，完成用户自定义Flink应用的执行。

以上就是主要的Flink On Yarn客户端作业的提交过程解析。这个提交过程相对来说还是比较复杂的，包含着很多部署配置参数，资源以及权限的校验和分配，ApplicationMaster的提交启动，并伴随AM启动后执行的一系列Flink服务初始化，以及我们关心的用户应用程序的调用入口，发现了在Application的部署模式下，用户应用程序的调用是在集群侧，也就是Leader DispatchRunner服务中完成的。

当然，DispatchRunner服务负责的任务远不止于此，上述流程中还有更多的细节等待我们去挖掘和学习，这篇文章可能只是让我们对提交流程有了一个初步的大体认识，对于更多深入的部分，需要我们不断思考不断挖掘，也欢迎大家交流观点和看法，感谢！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/65061.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Backend - C# 的日志 NLog日志

Backend - C# 的日志 NLog日志

目录 一、注入依赖和使用 logger 二、配置记录文件 1.安装插件 NLog 2.创建 nlog.config 配置文件 3. Programs配置日志信息 4. 设置 appsettings.json 的 LogLevel 5. 日志设定文件和日志级别的优先级 &#xff08;1&#xff09;常见的日志级别优先级 &#xff08;2&…
阅读更多...
ESP32自动下载电路分享

ESP32自动下载电路分享

下面是一个ESP32系列或者ESP8266等电路的一个自动下载电路 在ESP32等模块需要烧写程序的时候&#xff0c;需要通过将EN引脚更改为低电平并将IO0引脚设置为低电平来切换到烧写模式。 有时候也会采用先将IO接到一个按键上&#xff0c;按住按键拉低IO0的同时重新上电的方式进入烧写…
阅读更多...
QML自定义数值编辑框SpinBox样式

QML自定义数值编辑框SpinBox样式

代码展示 import QtQuick 2.9 import QtQuick.Window 2.2 import QtQuick.Controls 2.1Window {visible: truewidth: 640height: 480title: qsTr("Hello World")SpinBox {id: controlvalue: 50editable: truecontentItem: TextInput {z: 2text: control.textFromVal…
阅读更多...
魅族手机调用tts失败解决

魅族手机调用tts失败解决

安装了阅读、MultiTTS之后&#xff0c;发现阅读的时候一直tts初始化失败&#xff0c;换了多个tts软件也不行。。。 解决方法&#xff1a;tts软件设置后台运行权限 打开“手机管家”权限管理后台管理找到自己安装的tts软件&#xff08;比如我是MultiTTS&#xff09;&#xff0c…
阅读更多...
1-markdown转网页样式页面 --[制作网页模板] 【测试代码下载】

1-markdown转网页样式页面 --[制作网页模板] 【测试代码下载】

markdown转网页 将Markdown转换为带有样式的网页页面通常涉及以下几个步骤&#xff1a;首先&#xff0c;需要使用Markdown解析器将Markdown文本转换为HTML&#xff1b;其次&#xff0c;应用CSS样式来美化HTML内容。此外&#xff0c;还可以加入JavaScript以增加交互性。下面我将…
阅读更多...
Eplan 项目结构(高层代号、安装地点、位置代号)

Eplan 项目结构(高层代号、安装地点、位置代号)

Eplan中的项目结构分为3个层次&#xff1a; &#xff08;1&#xff09;功能面结构。指明这个系统的功能&#xff0c;有什么用途。在EPlan中&#xff0c;指的就是"高层代号&#xff08;&#xff09;"。 一般指的是线体。 &#xff08;2&#xff09;位置面结构。指明该…
阅读更多...
《Armv8-A virtualization》学习笔记

《Armv8-A virtualization》学习笔记

1.MAIR 的全称是 Memory Attribute Indirection Register。它是ARM架构中的一种寄存器&#xff0c;用于定义内存的属性&#xff0c;并提供一种间接访问内存属性的机制。MAIR寄存器包含多个字段&#xff0c;这些字段指示不同类型内存的属性&#xff0c;例如是否可以缓存、是否为…
阅读更多...
NLP 复习大纲

NLP 复习大纲

CH3 激活函数意义 增强网络表达能力&#xff0c;引入非线性因素 连续可导的非线性函数 尽可能简单 导数的值域要在合适的范围内 为什么会发生梯度消失 误差传播的迭代公式为&#xff1a; 其中需要用到激活函数的导数&#xff0c;而激活函数的导数值小于1时&#xff0c;误差经过…
阅读更多...
如何使用OBS Studio录制屏幕？

如何使用OBS Studio录制屏幕？

可以进入官网或github进行下载&#xff1a; https://obsproject.com/download 安装包解压后进入bin 进入64-bit 选择obs 64 进入OBS Studio后在来源内右键&#xff0c;选择添加 选择添加显示器采集即可录取整个屏幕&#xff0c;窗口采集可选择窗口进行录制 选择对应显示器即配置…
阅读更多...
深入理解连接池：从数据库到HTTP的优化之道

深入理解连接池：从数据库到HTTP的优化之道

在现代应用开发中&#xff0c;高效的资源管理是关键&#xff0c;其中连接池&#xff08;Connection Pool&#xff09;技术起到了至关重要的作用。本文将带你深入了解连接池的概念及其在数据库和HTTP通信中的应用&#xff0c;结合 JDBC 与 Druid 的关系&#xff0c;以及 HttpURL…
阅读更多...
XIAO Esp32 S3 网络摄像头——3音视频监控

XIAO Esp32 S3 网络摄像头——3音视频监控

1、介绍 之前分别介绍了音频和视频的接收，本文是整合了前2篇文章，实现了音视频的同时获取。 效果： 用xiao esp35 s3自制一个网络摄像头 2、适用场景广泛 家庭安防 无论是门前监控，还是室内安全，自制摄像头可以让你轻松把握每个角落，实时查看视频流，防止任何潜在风险。…
阅读更多...
9.类的定义与使用

9.类的定义与使用

类的定义构造函数(__init__)实例变量类变量方法(实例方法)类方法(classmethod)静态方法(staticmethod)属性装饰器(property)私有属性与方法继承多态方法重写super()函数类的文档字符串类的属性和方法访问控制 1.类的定义: 如int,list,tuple等等都是类,还可以通过class方法自己…
阅读更多...
【文献精读笔记】Explainability for Large Language Models: A Survey （大语言模型的可解释性综述）（三）

【文献精读笔记】Explainability for Large Language Models: A Survey （大语言模型的可解释性综述）（三）

****非斜体正文为原文献内容&#xff08;也包含笔者的补充&#xff09;&#xff0c;灰色块中是对文章细节的进一步详细解释&#xff01; 3.2 全局解释&#xff08;Global Explanation&#xff09; 与旨在解释模型个体预测的局部解释不同&#xff0c;全局解释提供了对语言模型…
阅读更多...
使用消息队列可能遇到的问题及其解决办法

使用消息队列可能遇到的问题及其解决办法

在使用消息队列时&#xff0c;可能会遇到以下三个问题&#xff1a; 一.消息丢失 产生的原因&#xff1a;消息发送出去&#xff0c;由于网络问题或系统异常没有抵达服务器&#xff1b; 解决办法&#xff1a; 做好容错方法&#xff08;try-catch&#xff09;&#xff0c;发送…
阅读更多...
学习随笔：word2vec在win11 vs2022下编译、测试运行

学习随笔：word2vec在win11 vs2022下编译、测试运行

word2vec 官网word2vec的本质是在自然语言词条数据集与计算机浮点数据集之间建立双射关系。word2vec建立的数据集最厉害的一点是&#xff0c;将自然语言词条数据集内部的推理过程&#xff0c;映射到了计算机浮点数据集内部的数值运算。我个人感觉理解这个数据映射方式是理解AI大…
阅读更多...
vue cli更新遇到的问题(vue -V查询版本号不变的问题)

vue cli更新遇到的问题(vue -V查询版本号不变的问题)

1.镜像地址选择 npm会去默认的registry远程仓库中下载指定内容 该过程可能十分缓慢 因此我们可以切换默认仓库为镜像地址 npm config set registry https://registry.npmmirror.com 通过该指令可以从最新的镜像地址下载指定内容(镜像地址可能会有变 有变请重新查询) 2.下载 …
阅读更多...
qt中如何判断字符串是否为数字，整数，浮点数？

qt中如何判断字符串是否为数字，整数，浮点数？

在 Qt 中&#xff0c;可以使用多种方法来判断字符串是否为数字、整数或浮点数。Qt 提供了一些方便的字符串和数值处理函数&#xff0c;可以帮助你实现这些判断。以下是几种常见的方法&#xff1a; 1. 使用 QRegularExpression Qt 提供了 QRegularExpression 类&#xff0c;可…
阅读更多...
搭建nginx文件服务器

搭建nginx文件服务器

方法一&#xff1a;通过docker方式搭建 1、创建一个nginx配置文件/etc/nginx/nginx.conf user nginx; worker_processes 1;error_log /var/log/nginx/error.log warn; pid /var/run/nginx.pid;events {worker_connections 1024; }http {include mime.types;default_typ…
阅读更多...
练习题：37

练习题：37

目录 Python题目 题目 题目分析 套接字概念剖析 通信原理分析 服务器 - 客户端连接建立过程&#xff1a; 基于套接字通信的底层机制&#xff1a; 代码实现 基于 TCP 的简单服务器 - 客户端通信示例 服务器端代码&#xff08;tcp_server.py&#xff09; 客户端代码&a…
阅读更多...
2017年IMO几何预选题第7题

2017年IMO几何预选题第7题

凸四边形 A B C D ABCD ABCD 有内切圆 I I I, △ D A B \triangle DAB △DAB, △ A B C \triangle ABC △ABC, △ B C D \triangle BCD △BCD, △ C D A \triangle CDA △CDA 的内心分别为 I a I_a Ia​, I b I_b Ib​, I c I_c Ic​, I d I_d Id​. △ A I b I d \…
阅读更多...
最新文章

Copyright @ 2022~2023