前言

本篇我们从典型Flink用户代码入口开始分析整个Flink作业的执行流程。
我们以wordcount为例：

  1. public static void main(String[] args) throws Exception {
  3.         // Checking input parameters
  4.         final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
  6.         // set up the execution environment
  7.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  9.         // make parameters available in the web interface
  10.         env.getConfig().setGlobalJobParameters(params);
  12.         // get input data
  13.         DataStream<String> text = null;
  14.         if (params.has("input")) {
  15.             // union all the inputs from text files
  16.             for (String input : params.getMultiParameterRequired("input")) {
  17.                 if (text == null) {
  18.                     text = env.readTextFile(input);
  19.                 } else {
  20.                     text = text.union(env.readTextFile(input));
  21.                 }
  22.             }
  23.             Preconditions.checkNotNull(text, "Input DataStream should not be null.");
  24.         } else {
  25.             System.out.println("Executing WordCount example with default input data set.");
  26.             System.out.println("Use --input to specify file input.");
  27.             // get default test text data
  28.             text = env.fromElements(WordCountData.WORDS);
  29.         }
  31.         DataStream<Tuple2<String, Integer>> counts =
  32.             // split up the lines in pairs (2-tuples) containing: (word,1)
  33.             text.flatMap(new Tokenizer())
  34.             // group by the tuple field "0" and sum up tuple field "1"
  35.             .keyBy(value -> value.f0).sum(1);
  37.         // emit result
  38.         if (params.has("output")) {
  39.             counts.writeAsText(params.get("output"));
  40.         } else {
  41.             System.out.println("Printing result to stdout. Use --output to specify output path.");
  42.             counts.print();
  43.         }
  44.         // execute program
  45.         env.execute("Streaming WordCount");
  46.     }

创建ExecutionEnvironment

Flink用户代码执行的第一步通常为获取执行的environment。

  1. final StreamExecutionEnvironment env = 
  2.     StreamExecutionEnvironment.getExecutionEnvironment();

那我们开始从getExecutionEnvironment方法开始分析

  1. public static StreamExecutionEnvironment getExecutionEnvironment() {
  2.     return getExecutionEnvironment(new Configuration());
  3. }
  5. // 进一步调用 getExecutionEnvironment(Configuration configuration)
  6. public static StreamExecutionEnvironment getExecutionEnvironment(Configuration configuration) {
  7.     return Utils.resolveFactory(threadLocalContextEnvironmentFactory, contextEnvironmentFactory)
  8.         .map(factory -> factory.createExecutionEnvironment(configuration))  // 创建默认的执行环境
  9.         .orElseGet(() -> StreamExecutionEnvironment.createLocalEnvironment(configuration)); // 创建local执行环境
  10. }

其中Util.resolveFactory方法接收两个参数，第一个是位于ThreadLocal中的factory，另一个是静态Factory。如果Threadlocal中存在一个factory，返回这个factory，否则返回静态factory。
如果从这两个factory中都无法创建出ExecutionEnvironment，则调用StreamExecutionEnvironment.createLocalEnvironment(configuration)方法。会创建出一个LocalStreamEnvironment，意味着使用本地运行模式，所有的任务在同一个JVM中运行。

如果用户使用命令行方式将Flink作业提交到集群，可参考03-flink源码任务提交流程分析中，程序会执行如下逻辑：

  1.   - CliFrontend.executeProgram
  2.   - ClientUtils.executeProgram
  3.   - StreamContextEnvironment.setAsContext
  4.   - StreamExecutionEnvironment.initializeContextEnvironment

通过这些调用，StreamExecutionEnvironmentFactory会被创建出来并设置到StreamExecutionEnvironment的threadLocalContextEnvironmentFactory和contextEnvironmentFactory中。调用这个factory的createExecutionEnvironment方法创建出一个StreamContextEnvironment对象。这是将作业提交到远程Flink分布式集群的运行环境。

execute方法执行作业

以上完成执行环境的创建，接下来分析execute方法启动Flink作业，代码如下：

  1. // 无参方法
  2. public JobExecutionResult execute() throws Exception {
  3.     // getJobName方法从配置文件pipeline.name配置项获取Job name
  4.     // 如果没有配置，使用默认名称"Flink Streaming Job"
  5.     return execute(getJobName());
  6. }
  8. // jobname执行方法
  9. public JobExecutionResult execute(String jobName) throws Exception {
  10.     Preconditions.checkNotNull(jobName, "Streaming Job name should not be null.");
  11.     return execute(getStreamGraph(jobName));
  12. }

在execute作业之前，有一个getStreamGraph方法，将作业转换为StreamGraph。续后会完成StreamGraph生成流程，接下来进一步分析execute方法。

接下来的execute方法的执行步骤随着ExecutionEnvironment的不同而不同。

StreamContextEnvironment的execute方法：

  1.     @Override
  2.     public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
  3.         // 异步执行作业
  4.         final JobClient jobClient = executeAsync(streamGraph);
  5.         // 获取配置的作业监听器
  6.         final List<JobListener> jobListeners = getJobListeners();
  8.         try {
  9.             // 获取作业执行结果，逐个通知作业监听器
  10.             final JobExecutionResult  jobExecutionResult = getJobExecutionResult(jobClient);
  11.             jobListeners.forEach(jobListener ->
  12.                     jobListener.onJobExecuted(jobExecutionResult, null));
  13.             return jobExecutionResult;
  14.         } catch (Throwable t) {
  15.             jobListeners.forEach(jobListener ->
  16.                     jobListener.onJobExecuted(null, ExceptionUtils.stripExecutionException(t)));
  17.             ExceptionUtils.rethrowException(t);
  19.             // never reached, only make javac happy
  20.             return null;
  21.         }
  22.     }

LocalStreamEnvironment的execute方法：

  1.     @Override
  2.     public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
  3.         return super.execute(streamGraph);
  4.     }

调用了父类的execute方法。它的父类正是StreamExecutionEnvironment。

  1. /**
  2.      * Triggers the program execution. The environment will execute all parts of
  3.      * the program that have resulted in a "sink" operation. Sink operations are
  4.      * for example printing results or forwarding them to a message queue.
  5.      *
  6.      * @param streamGraph the stream graph representing the transformations
  7.      * @return The result of the job execution, containing elapsed time and accumulators.
  8.      * @throws Exception which occurs during job execution.
  9.      */
  10.     @Internal
  11.     public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
  12.         // 异步执行作业
  13.         final JobClient jobClient = executeAsync(streamGraph);
  15.         try {
  16.             final JobExecutionResult jobExecutionResult;
  17.             // 使用attached模式执行作业由于需要保持client端不关闭，所以这里同步等待作业执行结果
  18.             if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
  19.                 jobExecutionResult = jobClient.getJobExecutionResult().get();
  20.             } else {
  21.                 // // 异步模式则不需要
  22.                 jobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
  23.             }
  24.             // 逐个通知jobListener
  25.             jobListeners.forEach(jobListener -> jobListener.onJobExecuted(jobExecutionResult, null));
  27.             return jobExecutionResult;
  28.         } catch (Throwable t) {
  29.             // get() on the JobExecutionResult Future will throw an ExecutionException. This
  30.             // behaviour was largely not there in Flink versions before the PipelineExecutor
  31.             // refactoring so we should strip that exception.
  32.             Throwable strippedException = ExceptionUtils.stripExecutionException(t);
  34.             jobListeners.forEach(jobListener -> {
  35.                 jobListener.onJobExecuted(null, strippedException);
  36.             });
  37.             ExceptionUtils.rethrowException(strippedException);
  39.             // never reached, only make javac happy
  40.             return null;
  41.         }
  42.     }

接下来我们分析StreamContextEnvironment的executeAsync方法。

  1.     @Override
  2.     public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
  3.         // 检查一个environment中不能调用多次execute或executeAsync
  4.         validateAllowedExecution();
  5.         // 调用父类的executeAsync方法
  6.         // 和LocalStreamEnvironment中的executeAsync相同
  7.         final JobClient jobClient = super.executeAsync(streamGraph);
  9.         if (!suppressSysout) {
  10.             System.out.println("Job has been submitted with JobID " + jobClient.getJobID());
  11.         }
  13.         return jobClient;
  14.     }

最终两个Environment的executeAsync方法归为一致。

下面是StreamExecutionEnvironment类的executeAsync方法：

  1. /**
  2.      * Triggers the program execution asynchronously. The environment will execute all parts of
  3.      * the program that have resulted in a "sink" operation. Sink operations are
  4.      * for example printing results or forwarding them to a message queue.
  5.      *
  6.      * @param streamGraph the stream graph representing the transformations
  7.      * @return A {@link JobClient} that can be used to communicate with the submitted job, completed on submission succeeded.
  8.      * @throws Exception which occurs during job execution.
  9.      */
  10.     @Internal
  11.     public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
  12.         // 检查streamGraph不能为null
  13.         checkNotNull(streamGraph, "StreamGraph cannot be null.");
  14.         // 检查部署目标配置不能为null
  15.         // 部署目标即作业运行的模式，例如本地模式，远程模式，yarn模式或者是k8s模式
  16.         checkNotNull(configuration.get(DeploymentOptions.TARGET), "No execution.target specified in your configuration file.");
  18.         // 基于configuration创建一个合适的执行器工厂类
  19.         // 获取作业执行器工厂类
  20.         final PipelineExecutorFactory executorFactory =
  21.             executorServiceLoader.getExecutorFactory(configuration);
  23.         checkNotNull(
  24.             executorFactory,
  25.             "Cannot find compatible factory for specified execution.target (=%s)",
  26.             configuration.get(DeploymentOptions.TARGET));
  28.         // 从执行器工厂获取执行器，运行包含用户作业的streamGraph
  29.         CompletableFuture<JobClient> jobClientFuture = executorFactory
  30.             .getExecutor(configuration) //根据configuration获取对应的执行期
  31.             .execute(streamGraph, configuration, userClassloader); // 根据stramgraph, configuration, 类加载器 开始执行作业
  33.         try {
  34.             // 通知各个作业监听器作业已提交
  35.             JobClient jobClient = jobClientFuture.get();
  36.             jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(jobClient, null));
  37.             return jobClient;
  38.         } catch (ExecutionException executionException) {
  39.             final Throwable strippedException = ExceptionUtils.stripExecutionException(executionException);
  40.             jobListeners.forEach(jobListener -> jobListener.onJobSubmitted(null, strippedException));
  42.             throw new FlinkException(
  43.                 String.format("Failed to execute job '%s'.", streamGraph.getJobName()),
  44.                 strippedException);
  45.         }
  46.     }

到这里不同的执行环境具体的逻辑又要开始走向不同。这些逻辑的分叉点在executorServiceLoader.getExecutorFactory(configuration)，不同环境下获取到的PipelineExecutorFactory是不同的。在创建LocalStreamEnvironment或者是StreamContextEnvironment的时候executorServiceLoader变量传入的是DefaultExecutorServiceLoader。我们查看它的getExecutorFactory方法。如下所示：

DefaultExecutorServiceLoader的getExecutorFactory方法：

  1. public PipelineExecutorFactory getExecutorFactory(final Configuration configuration) {
  2.         checkNotNull(configuration);
  4.         // 根据配置文件加载并实例化PipelineExecutorFactory的实现类
  5.         final ServiceLoader<PipelineExecutorFactory> loader =
  6.                 ServiceLoader.load(PipelineExecutorFactory.class);
  8.         final List<PipelineExecutorFactory> compatibleFactories = new ArrayList<>();
  9.         final Iterator<PipelineExecutorFactory> factories = loader.iterator();
  10.         while (factories.hasNext()) {
  11.             try {
  12.                 // 遍历所有加载的factory
  13.                 // 只保留和配置文件兼容的factory
  14.                 final PipelineExecutorFactory factory = factories.next();
  15.                 if (factory != null && factory.isCompatibleWith(configuration)) {
  16.                     compatibleFactories.add(factory);
  17.                 }
  18.             } catch (Throwable e) {
  19.                 if (e.getCause() instanceof NoClassDefFoundError) {
  20.                     LOG.info("Could not load factory due to missing dependencies.");
  21.                 } else {
  22.                     throw e;
  23.                 }
  24.             }
  25.         }
  27.         // 如果兼容的factory有多个，打印错误信息
  28.         if (compatibleFactories.size() > 1) {
  29.             final String configStr =
  30.                     configuration.toMap().entrySet().stream()
  31.                             .map(e -> e.getKey() + "=" + e.getValue())
  32.                             .collect(Collectors.joining("\n"));
  34.             throw new IllegalStateException("Multiple compatible client factories found for:\n" + configStr + ".");
  35.         }
  37.         // 如果兼容的factory是0个，抛出异常信息
  38.         if (compatibleFactories.isEmpty()) {
  39.             throw new IllegalStateException("No ExecutorFactory found to execute the application.");
  40.         }
  42.         // 返回这个唯一的factory
  43.         return compatibleFactories.get(0);
  44.     }

在这个方法中使用了Java SPI机制，根据META-INF/services内的配置文件动态加载并实例化PipelineExecutorFactory的子类。

PipelineExecutorFactory和PipelineExecutor

PipelineExecutorFactory

我们查找下org.apache.flink.core.execution.PipelineExecutorFactory配置文件，发现共有3个，分别位于源码flink-clients，flink-yarn和flink-kubernetes子项目中。下面逐个分析。

flink-clients中的org.apache.flink.core.execution.PipelineExecutorFactory内容如下：

  1. org.apache.flink.client.deployment.executors.RemoteExecutorFactory
  2. org.apache.flink.client.deployment.executors.LocalExecutorFactory

也就是说如果我们引入了flink-clients包，会创建RemoteExecutorFactory和LocalExecutorFactory实例。

我们在分别查看下他们的isCompatibleWith方法。

RemoteExecutorFactory的isCompatibleWith方法：

  1. @Override
  2. public boolean isCompatibleWith(final Configuration configuration) {
  3.     // 配置文件 中 execution.target=remote时；执行该方法
  4.     return RemoteExecutor.NAME.equalsIgnoreCase(configuration.get(DeploymentOptions.TARGET));
  5. }

要求配置文件中execution.target配置为remote，才会使用RemoteExecutorFactory。

LocalExecutorFactory的isCompatibleWith方法：

  1. @Override
  2. public boolean isCompatibleWith(final Configuration configuration) {
  3.     // 配置文件 中 execution.target=local 时；执行该方法
  4.     return LocalExecutor.NAME.equalsIgnoreCase(configuration.get(DeploymentOptions.TARGET));
  5. }

要求配置文件中execution.target配置为local，才会使用LocalExecutorFactory。

flink-yarn中的org.apache.flink.core.execution.PipelineExecutorFactory内容如下：

  1. org.apache.flink.yarn.executors.YarnJobClusterExecutorFactory
  2. org.apache.flink.yarn.executors.YarnSessionClusterExecutorFactory

和上面类似，这里只说明各个factory兼容的配置项，不再贴出isCompatibleWith方法代码。

  • YarnJobClusterExecutorFactory：要求execution.target为yarn-per-job
  • YarnSessionClusterExecutorFactory：要求execution.target为yarn-session

flink-kubernetes中的org.apache.flink.core.execution.PipelineExecutorFactory内容如下：

  1. org.apache.flink.kubernetes.executors.KubernetesSessionClusterExecutorFactory

其中KubernetesSessionClusterExecutorFactory要求execution.target配置为kubernetes-session。

PipelineExecutor

接下来我们重点讨论两个Executor：LocalExecutor和RemoteExecutor。这两个executor创建的逻辑比较简单，此处不再介绍。

PipelineExecutor执行作业的方法为execute。它有3个参数：

  • pipeline：要执行的作业，指的是StreamGraph。
  • configuration：作业的配置。
  • userCodeClassloader：用户作业的类加载器。和Flink本身使用不同类加载器的原因是不同用户作业加载的class可能冲突，用户作业和Flink框架本身加载的class也可能冲突。为了避免这种冲突，用户作业采用不同的类加载器加载。

LocalExecutor

LocalExecutor用于在本地执行任务。它的execute方法如下：

  1.     public CompletableFuture<JobClient> execute(Pipeline pipeline, Configuration configuration, 
  2.                                                 ClassLoader userCodeClassloader) throws Exception {
  3.         // 校验检查
  4.         checkNotNull(pipeline);
  5.         checkNotNull(configuration);
  7.         //
  8.         Configuration effectiveConfig = new Configuration();
  9.         effectiveConfig.addAll(this.configuration);
  10.         effectiveConfig.addAll(configuration);
  12.         // we only support attached execution with the local executor.
  13.         // 只支持ATTACHED模式运行
  14.         checkState(configuration.getBoolean(DeploymentOptions.ATTACHED));
  16.         // 将StreamGraph转换为JobGraph
  17.         final JobGraph jobGraph = getJobGraph(pipeline, effectiveConfig);
  19.         // 创建一个MiniCluster
  20.         // 并调用MiniCluster的submitJob方法，提交作业
  21.         return PerJobMiniClusterFactory.createWithFactory(effectiveConfig, miniClusterFactory)
  22.             .submitJob(jobGraph, userCodeClassloader);
  23.     }

在submit作业之前，需要将StreamGraph转换为JobGraph。后续会补充JobGraph如何生成。

在方法的最后调用了PerJobMiniClusterFactory的submitJob方法。PerJobMiniClusterFactory实际操作的是MiniCluster对象。顾名思义，它是一个”小型集群”，所有的作业都在本地运行。

PerJobMiniClusterFactory的submitJob方法代码如下：

  1. /**
  2.      * Starts a {@link MiniCluster} and submits a job.
  3.      */
  4.     public CompletableFuture<JobClient> submitJob(JobGraph jobGraph, ClassLoader userCodeClassloader) throws Exception {
  5.         // 获取MiniCluster的配置，指定最大并行度
  6.         MiniClusterConfiguration miniClusterConfig = getMiniClusterConfig(jobGraph.getMaximumParallelism());
  7.         // 创建出一个MiniCluster
  8.         MiniCluster miniCluster = miniClusterFactory.apply(miniClusterConfig);
  9.         // 启动miniCluster
  10.         miniCluster.start();
  12.         return miniCluster
  13.             .submitJob(jobGraph)
  14.             .thenApplyAsync(FunctionUtils.uncheckedFunction(submissionResult -> {
  15.                 org.apache.flink.client.ClientUtils.waitUntilJobInitializationFinished(
  16.                     () -> miniCluster.getJobStatus(submissionResult.getJobID()).get(),
  17.                     () -> miniCluster.requestJobResult(submissionResult.getJobID()).get(),
  18.                     userCodeClassloader);
  19.                 return submissionResult;
  20.             }))
  21.             .thenApply(result -> new MiniClusterJobClient(
  22.                     result.getJobID(),
  23.                     miniCluster,
  24.                     userCodeClassloader,
  25.                     MiniClusterJobClient.JobFinalizationBehavior.SHUTDOWN_CLUSTER))
  26.             .whenComplete((ignored, throwable) -> {
  27.                 if (throwable != null) {
  28.                     // We failed to create the JobClient and must shutdown to ensure cleanup.
  29.                     shutDownCluster(miniCluster);
  30.                 }
  31.             })
  32.             .thenApply(Function.identity());
  33.     }

RemoteExecutor

remoteExecutor的execute方法位于它的父类AbstractSessionClusterExecutor中。代码和分析如下所示：

  1.     @Override
  2.     public CompletableFuture<JobClient> execute(@Nonnull final Pipeline pipeline,
  3.                                                 @Nonnull final Configuration configuration, 
  4.                                                 @Nonnull final ClassLoader userCodeClassloader) throws Exception {
  5.         // 和之前相同，仍然是生成JobGraph
  6.         final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
  8.         try (final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory
  9.              .createClusterDescriptor(configuration)) {
  10.             // 获取远程集群ID
  11.             final ClusterID clusterID = clusterClientFactory.getClusterId(configuration);
  12.             checkState(clusterID != null);
  14.             final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor.retrieve(clusterID);
  15.             // 创建出clusterClient，用户和远程集群通信，提交作业
  16.             ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();
  17.             return clusterClient
  18.                     .submitJob(jobGraph)  // 提交作业
  19.                     .thenApplyAsync(FunctionUtils.uncheckedFunction(jobId -> {
  20.                         ClientUtils.waitUntilJobInitializationFinished(
  21.                             () -> clusterClient.getJobStatus(jobId).get(),
  22.                             () -> clusterClient.requestJobResult(jobId).get(),
  23.                             userCodeClassloader);
  24.                         return jobId;
  25.                     }))
  26.                     .thenApplyAsync(jobID -> (JobClient) new ClusterClientJobClientAdapter<>(
  27.                             clusterClientProvider,
  28.                             jobID,
  29.                             userCodeClassloader))
  30.                     .whenComplete((ignored1, ignored2) -> clusterClient.close());
  31.         }
  32.     }

ClusterClient根据集群类型的不同，有两个子类：MiniClusterClient和RestClusterClient。其中MiniClusterClient用户和MiniCluster通信，它的submitJob方法实际调用的是MiniCluster的submitJob方法。

RestClusterClient通过http rest请求和远程集群通信。它的submitJob方法如下所示：

  1. @Override
  2.     public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
  3.         CompletableFuture<java.nio.file.Path> jobGraphFileFuture = CompletableFuture.supplyAsync(() -> {
  4.             try {
  5.                 // 创建临时文件
  6.                 final java.nio.file.Path jobGraphFile = Files.createTempFile("flink-jobgraph", ".bin");
  7.                 // 写到输出对象流中
  8.                 try (ObjectOutputStream objectOut = new ObjectOutputStream(Files.newOutputStream(jobGraphFile))) {
  9.                     objectOut.writeObject(jobGraph);
  10.                 }
  11.                 // 返回文件路径
  12.                 return jobGraphFile;
  13.             } catch (IOException e) {
  14.                 throw new CompletionException(new FlinkException("Failed to serialize JobGraph.", e));
  15.             }
  16.         }, executorService);
  18.         // 在JobGraph写入文件完成之后执行
  19.         CompletableFuture<Tuple2<JobSubmitRequestBody, Collection<FileUpload>>> requestFuture = jobGraphFileFuture.thenApply(jobGraphFile -> {
  20.             // jar文件名列表
  21.             List<String> jarFileNames = new ArrayList<>(8);
  22.             // 缓存的分布式文件列表
  23.             List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames = new ArrayList<>(8);
  24.             // 需要上传的文件列表
  25.             Collection<FileUpload> filesToUpload = new ArrayList<>(8);
  26.             //首先加载以及序列化的jobgraph文件
  27.             filesToUpload.add(new FileUpload(jobGraphFile, RestConstants.CONTENT_TYPE_BINARY));
  29.             // 从JobGraph中获取用户的jar文件路径，加入到上传列表
  30.             for (Path jar : jobGraph.getUserJars()) {
  31.                 jarFileNames.add(jar.getName());
  32.                 filesToUpload.add(new FileUpload(Paths.get(jar.toUri()), RestConstants.CONTENT_TYPE_JAR));
  33.             }
  36.             for (Map.Entry<String, DistributedCache.DistributedCacheEntry> artifacts : jobGraph.getUserArtifacts().entrySet()) {
  37.                 final Path artifactFilePath = new Path(artifacts.getValue().filePath);
  38.                 try {
  39.                     // Only local artifacts need to be uploaded.
  40.                     // 上传用户作业运行所需的其它类型文件
  41.                     // 只需要上传本地储存的文件
  42.                     if (!artifactFilePath.getFileSystem().isDistributedFS()) {
  43.                         artifactFileNames.add(new JobSubmitRequestBody.DistributedCacheFile(artifacts.getKey(), artifactFilePath.getName()));
  44.                         filesToUpload.add(new FileUpload(Paths.get(artifacts.getValue().filePath), RestConstants.CONTENT_TYPE_BINARY));
  45.                     }
  46.                 } catch (IOException e) {
  47.                     throw new CompletionException(
  48.                         new FlinkException("Failed to get the FileSystem of artifact " + artifactFilePath + ".", e));
  49.                 }
  50.             }
  51.             // 创建作业提交请求体
  52.             final JobSubmitRequestBody requestBody = new JobSubmitRequestBody(
  53.                 jobGraphFile.getFileName().toString(),
  54.                 jarFileNames,
  55.                 artifactFileNames);
  57.             // 返回请求体和需要上传的文件
  58.             return Tuple2.of(requestBody, Collections.unmodifiableCollection(filesToUpload));
  59.         });
  61.         // 请求构建完毕后提交请求
  62.         final CompletableFuture<JobSubmitResponseBody> submissionFuture = requestFuture.thenCompose(
  63.             requestAndFileUploads -> sendRetriableRequest(
  64.                 JobSubmitHeaders.getInstance(),
  65.                 EmptyMessageParameters.getInstance(),
  66.                 requestAndFileUploads.f0,
  67.                 requestAndFileUploads.f1,
  68.                 isConnectionProblemOrServiceUnavailable())
  69.         );
  71.         // 请求发送完毕之后，删除JobGraph临时文件
  72.         submissionFuture
  73.             .thenCombine(jobGraphFileFuture, (ignored, jobGraphFile) -> jobGraphFile)
  74.             .thenAccept(jobGraphFile -> {
  75.             try {
  76.                 Files.delete(jobGraphFile);
  77.             } catch (IOException e) {
  78.                 LOG.warn("Could not delete temporary file {}.", jobGraphFile, e);
  79.             }
  80.         });
  82.         return submissionFuture
  83.             .thenApply(ignore -> jobGraph.getJobID())
  84.             .exceptionally(
  85.                 (Throwable throwable) -> {
  86.                     throw new CompletionException(new JobSubmissionException(jobGraph.getJobID(),
  87.                         "Failed to submit JobGraph.", ExceptionUtils.stripCompletionException(throwable)));
  88.                 });
  89.     }

由于远程集群的JobManager和TaskManager需要单独部署启动，和本篇内容关联不大，此处不再详细描述。

MiniCluster

前面已经简单介绍过MiniCluster，它用于在本地环境执行作业。

start方法

启动MiniCluster的逻辑位于start方法中。

  1.     /**
  2.      * Starts the mini cluster, based on the configured properties.
  3.      *
  4.      * @throws Exception This method passes on any exception that occurs during the startup of
  5.      *                   the mini cluster.
  6.      */
  7.     public void start() throws Exception {
  8.         synchronized (lock) {
  9.             checkState(!running, "MiniCluster is already running");
  11.             LOG.info("Starting Flink Mini Cluster");
  12.             LOG.debug("Using configuration {}", miniClusterConfiguration);
  14.             final Configuration configuration = miniClusterConfiguration.getConfiguration();
  15.             // MiniCluster中的组件使用同一个共享的RPC服务
  16.             final boolean useSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED;
  18.             try {
  19.                 // 初始化IO相关配置，是否复写和是否创建output directory等
  20.                 initializeIOFormatClasses(configuration);
  22.                 LOG.info("Starting Metrics Registry");
  23.                 // 创建监控相关配置
  24.                 metricRegistry = createMetricRegistry(configuration);
  26.                 // bring up all the RPC services
  27.                 LOG.info("Starting RPC Service(s)");
  29.                 final RpcServiceFactory dispatcherResourceManagerComponentRpcServiceFactory;
  30.                 final RpcService metricQueryServiceRpcService;
  32.                 // 使用共享rpc服务
  33.                 if (useSingleRpcService) {
  34.                     // we always need the 'commonRpcService' for auxiliary calls
  35.                     // 创建本地RPC服务
  36.                     commonRpcService = createLocalRpcService(configuration);
  37.                     // 创建通用RPC服务工厂
  38.                     final CommonRpcServiceFactory commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
  39.                     // TaskManager RPC服务工厂使用RPC服务工厂
  40.                     taskManagerRpcServiceFactory = commonRpcServiceFactory;
  41.                     dispatcherResourceManagerComponentRpcServiceFactory = commonRpcServiceFactory;
  43.                     // 启动RPC查询服务
  44.                     metricQueryServiceRpcService = MetricUtils.startLocalMetricsRpcService(configuration);
  45.                 } else {
  47.                     // start a new service per component, possibly with custom bind addresses
  48.                     // 如果不共用RPC服务，获取JobManager和TaskManager地址和端口范围
  49.                     final String jobManagerExternalAddress = miniClusterConfiguration.getJobManagerExternalAddress();
  50.                     final String taskManagerExternalAddress = miniClusterConfiguration.getTaskManagerExternalAddress();
  51.                     final String jobManagerExternalPortRange = miniClusterConfiguration.getJobManagerExternalPortRange();
  52.                     final String taskManagerExternalPortRange = miniClusterConfiguration.getTaskManagerExternalPortRange();
  53.                     final String jobManagerBindAddress = miniClusterConfiguration.getJobManagerBindAddress();
  54.                     final String taskManagerBindAddress = miniClusterConfiguration.getTaskManagerBindAddress();
  56.                     // 分别创建各个组件的factory和服务等
  57.                     dispatcherResourceManagerComponentRpcServiceFactory =
  58.                         new DedicatedRpcServiceFactory(
  59.                             configuration,
  60.                             jobManagerExternalAddress,
  61.                             jobManagerExternalPortRange,
  62.                             jobManagerBindAddress);
  63.                     taskManagerRpcServiceFactory =
  64.                         new DedicatedRpcServiceFactory(
  65.                             configuration,
  66.                             taskManagerExternalAddress,
  67.                             taskManagerExternalPortRange,
  68.                             taskManagerBindAddress);
  70.                     // we always need the 'commonRpcService' for auxiliary calls
  71.                     // bind to the JobManager address with port 0
  72.                     commonRpcService = createRemoteRpcService(configuration, jobManagerBindAddress, 0);
  73.                     metricQueryServiceRpcService = MetricUtils.startRemoteMetricsRpcService(
  74.                         configuration,
  75.                         commonRpcService.getAddress());
  76.                 }
  78.                 // 启动监控查询服务
  79.                 metricRegistry.startQueryService(metricQueryServiceRpcService, null);
  81.                 // 创建进程监控指标组
  82.                 processMetricGroup = MetricUtils.instantiateProcessMetricGroup(
  83.                     metricRegistry,
  84.                     RpcUtils.getHostname(commonRpcService),
  85.                     ConfigurationUtils.getSystemResourceMetricsProbingInterval(configuration));
  87.                 // 创建IO线程池
  88.                 ioExecutor = Executors.newFixedThreadPool(
  89.                     ClusterEntrypointUtils.getPoolSize(configuration),
  90.                     new ExecutorThreadFactory("mini-cluster-io"));
  91.                 // 创建高可用服务
  92.                 haServices = createHighAvailabilityServices(configuration, ioExecutor);
  94.                 // 启动blobServer
  95.                 blobServer = new BlobServer(configuration, haServices.createBlobStore());
  96.                 blobServer.start();
  98.                 // 创建心跳服务
  99.                 heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
  101.                 // 创建blob缓存服务
  102.                 blobCacheService = new BlobCacheService(
  103.                     configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())
  104.                 );
  106.                 // 启动TaskManager
  107.                 startTaskManagers();
  109.                 // 创建监控查询获取服务
  110.                 MetricQueryServiceRetriever metricQueryServiceRetriever = new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService());
  112.                 // 创建Dispatcher和ResourceManager，它们在同一个进程中运行
  113.                 setupDispatcherResourceManagerComponents(configuration, dispatcherResourceManagerComponentRpcServiceFactory, metricQueryServiceRetriever);
  115.                 // 创建ResourceManager leader获取服务
  116.                 resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();
  117.                 dispatcherLeaderRetriever = haServices.getDispatcherLeaderRetriever();
  118.                 clusterRestEndpointLeaderRetrievalService = haServices.getClusterRestEndpointLeaderRetriever();
  119.                 // 创建Dispatcher gateway获取服务
  120.                 dispatcherGatewayRetriever = new RpcGatewayRetriever<>(
  121.                     commonRpcService,
  122.                     DispatcherGateway.class,
  123.                     DispatcherId::fromUuid,
  124.                     new ExponentialBackoffRetryStrategy(21, Duration.ofMillis(5L), Duration.ofMillis(20L)));
  125.                 resourceManagerGatewayRetriever = new RpcGatewayRetriever<>(
  126.                     commonRpcService,
  127.                     ResourceManagerGateway.class,
  128.                     ResourceManagerId::fromUuid,
  129.                     new ExponentialBackoffRetryStrategy(21, Duration.ofMillis(5L), Duration.ofMillis(20L)));
  130.                 // 创建WebMonitor leader获取服务
  131.                 webMonitorLeaderRetriever = new LeaderRetriever();
  133.                 // 分别启动这些服务
  134.                 resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);
  135.                 dispatcherLeaderRetriever.start(dispatcherGatewayRetriever);
  136.                 clusterRestEndpointLeaderRetrievalService.start(webMonitorLeaderRetriever);
  137.             }
  138.             catch (Exception e) {
  139.                 // cleanup everything
  140.                 try {
  141.                     close();
  142.                 } catch (Exception ee) {
  143.                     e.addSuppressed(ee);
  144.                 }
  145.                 throw e;
  146.             }
  148.             // create a new termination future
  149.             terminationFuture = new CompletableFuture<>();
  151.             // now officially mark this as running
  152.             running = true;
  154.             LOG.info("Flink Mini Cluster started successfully");
  155.         }
  156.     }

通过上面的分析我们熟悉了MiniCluster的启动流程。接下来分析下启动TaskManager的逻辑，位于startTaskManagers方法。

  1. @GuardedBy("lock")
  2. private void startTaskManagers() throws Exception {
  3.     final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers();
  4.     // 启动task manager的个数
  5.     LOG.info("Starting {} TaskManger(s)", numTaskManagers);
  7.     for (int i = 0; i < numTaskManagers; i++) {
  8.         startTaskExecutor();
  9.     }
  10. }

继续跟踪startTaskExecutor方法：

  1. @VisibleForTesting
  2. void startTaskExecutor() throws Exception {
  3.     synchronized (lock) {
  4.         final Configuration configuration = miniClusterConfiguration.getConfiguration();
  6.         // 创建TaskExecutor
  7.         final TaskExecutor taskExecutor = TaskManagerRunner.startTaskManager(
  8.             configuration,
  9.             new ResourceID(UUID.randomUUID().toString()),
  10.             taskManagerRpcServiceFactory.createRpcService(),
  11.             haServices,
  12.             heartbeatServices,
  13.             metricRegistry,
  14.             blobCacheService,
  15.             useLocalCommunication(),
  16.             ExternalResourceInfoProvider.NO_EXTERNAL_RESOURCES,
  17.             taskManagerTerminatingFatalErrorHandlerFactory.create(taskManagers.size()));
  19.         //  启动TaskExecutor
  20.         taskExecutor.start();
  21.         taskManagers.add(taskExecutor);
  22.     }
  23. }

taskmanger创建了多个taskexecutor,到这里TaskManager已经启动完毕。

Dispatcher

Dispatcher负责提交作业和创建出JobManager。

Dispatcher有2个子类，MiniDispatcher和StandaloneDispatcher。分别用于提交任务给MiniCluster和其他独立的cluster。其中StandaloneDispatcher的实现
最为简单，没有override父类Dispatcher的任何方法。

接下来我们分析下MiniDispatcher的submitJob方法。

  1. @Override
  2. public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
  3.     // 调用Dispatcher的submitJob方法
  4.     final CompletableFuture<Acknowledge> acknowledgeCompletableFuture = super.submitJob(jobGraph, timeout);
  6.     acknowledgeCompletableFuture.whenComplete(
  7.         (Acknowledge ignored, Throwable throwable) -> {
  8.             if (throwable != null) {
  9.                 onFatalError(new FlinkException(
  10.                     "Failed to submit job " + jobGraph.getJobID() + " in job mode.",
  11.                     throwable));
  12.             }
  13.         });
  15.     return acknowledgeCompletableFuture;
  16. }

继续查看父类Dispacher的submitJob方法，如下所示：

  1. @Override
  2. public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
  3.     log.info("Received JobGraph submission {} ({}).", jobGraph.getJobID(), jobGraph.getName());
  5.     try {
  6.         // 检查作业是否重复
  7.         if (isDuplicateJob(jobGraph.getJobID())) {
  8.             return FutureUtils.completedExceptionally(
  9.                 new DuplicateJobSubmissionException(jobGraph.getJobID()));
  10.         // 检查部分资源是否已配置
  11.         } else if (isPartialResourceConfigured(jobGraph)) {
  12.             return FutureUtils.completedExceptionally(
  13.                 new JobSubmissionException(jobGraph.getJobID(), "Currently jobs is not supported if parts of the vertices have " +
  14.                         "resources configured. The limitation will be removed in future versions."));
  15.         } else {
  16.             // 内部提交作业
  17.             return internalSubmitJob(jobGraph);
  18.         }
  19.     } catch (FlinkException e) {
  20.         return FutureUtils.completedExceptionally(e);
  21.     }
  22. }

接下来流程到了internalSubmitJob方法，代码如下：

  1. private CompletableFuture<Acknowledge> internalSubmitJob(JobGraph jobGraph) {
  2.     log.info("Submitting job {} ({}).", jobGraph.getJobID(), jobGraph.getName());
  3.     // 调用persistAndRunJob方法
  4.     final CompletableFuture<Acknowledge> persistAndRunFuture = waitForTerminatingJob(jobGraph.getJobID(), jobGraph,
  5.         this::persistAndRunJob)
  6.         .thenApply(ignored -> Acknowledge.get());
  8.     return persistAndRunFuture.handleAsync((acknowledge, throwable) -> {
  9.         if (throwable != null) {
  10.             cleanUpJobData(jobGraph.getJobID(), true);
  12.             ClusterEntryPointExceptionUtils.tryEnrichClusterEntryPointError(throwable);
  13.             final Throwable strippedThrowable = ExceptionUtils.stripCompletionException(throwable);
  14.             log.error("Failed to submit job {}.", jobGraph.getJobID(), strippedThrowable);
  15.             throw new CompletionException(
  16.                 new JobSubmissionException(jobGraph.getJobID(), "Failed to submit job.", strippedThrowable));
  17.         } else {
  18.             return acknowledge;
  19.         }
  20.     }, ioExecutor);
  21. }

继续跟踪persistAndRunJob方法：

  1. private void persistAndRunJob(JobGraph jobGraph) throws Exception {
  2.     // 存储JobGraph
  3.     jobGraphWriter.putJobGraph(jobGraph);
  4.     // // 调用runJob方法
  5.     runJob(jobGraph, ExecutionType.SUBMISSION);
  6. }

分析下runJob方法：

  1. private void runJob(JobGraph jobGraph, ExecutionType executionType) {
  2.     Preconditions.checkState(!runningJobs.containsKey(jobGraph.getJobID()));
  3.     // 作业时间戳
  4.     long initializationTimestamp = System.currentTimeMillis();
  5.     // 调用启动JobManager逻辑
  6.     CompletableFuture<JobManagerRunner> jobManagerRunnerFuture = createJobManagerRunner(jobGraph, initializationTimestamp);
  8.     DispatcherJob dispatcherJob = DispatcherJob.createFor(
  9.             jobManagerRunnerFuture,
  10.             jobGraph.getJobID(),
  11.             jobGraph.getName(),
  12.             initializationTimestamp);
  13.     // 将当前Job存入runningJobs集合
  14.     runningJobs.put(jobGraph.getJobID(), dispatcherJob);
  16.     final JobID jobId = jobGraph.getJobID();
  17.     // 处理Job提交结果，进行cleanup操作
  18.     final CompletableFuture<CleanupJobState> cleanupJobStateFuture = dispatcherJob.getResultFuture().handleAsync(
  19.         (dispatcherJobResult, throwable) -> {
  20.             Preconditions.checkState(runningJobs.get(jobId) == dispatcherJob, "The job entry in runningJobs must be bound to the lifetime of the DispatcherJob.");
  22.             if (dispatcherJobResult != null) {
  23.                 return handleDispatcherJobResult(jobId, dispatcherJobResult, executionType);
  24.             } else {
  25.                 return dispatcherJobFailed(jobId, throwable);
  26.             }
  27.         }, getMainThreadExecutor());
  29.     // 在cleanup完成之后终止作业
  30.     final CompletableFuture<Void> jobTerminationFuture = cleanupJobStateFuture
  31.         .thenApply(cleanupJobState -> removeJob(jobId, cleanupJobState))
  32.         .thenCompose(Function.identity());
  34.     FutureUtils.assertNoException(jobTerminationFuture);
  35.     registerDispatcherJobTerminationFuture(jobId, jobTerminationFuture);
  36. }

这里到了MiniCluster提交作业流程的最后一步，创建JobManagerRunner。

createJobManagerRunner方法分析如下：

  1. CompletableFuture<JobManagerRunner> createJobManagerRunner(JobGraph jobGraph, long initializationTimestamp) {
  2.     // 获取RPC服务
  3.     final RpcService rpcService = getRpcService();
  5.     return CompletableFuture.supplyAsync(
  6.         () -> {
  7.             try {
  8.                 // 创建出一个JobManager启动器
  9.                 // 传入JobGraph和其他参数
  10.                 JobManagerRunner runner = jobManagerRunnerFactory.createJobManagerRunner(
  11.                     jobGraph,
  12.                     configuration,
  13.                     rpcService,
  14.                     highAvailabilityServices,
  15.                     heartbeatServices,
  16.                     jobManagerSharedServices,
  17.                     new DefaultJobManagerJobMetricGroupFactory(jobManagerMetricGroup),
  18.                     fatalErrorHandler,
  19.                     initializationTimestamp);
  20.                 // 启动JobManager启动器
  21.                 runner.start();
  22.                 return runner;
  23.             } catch (Exception e) {
  24.                 throw new CompletionException(new JobInitializationException(jobGraph.getJobID(), "Could not instantiate JobManager.", e));
  25.             }
  26.         },
  27.         ioExecutor); // do not use main thread executor. Otherwise, Dispatcher is blocked on JobManager creation
  28. }

下面我们开始分析JobManager启动逻辑。

JobManagerRunner和JobManagerRunnerFactory

JobManagerRunnerFactory

接下来我们分析JobManagerRunnerFactory的唯一实现类DefaultJobManagerRunnerFactory的createJobManagerRunner方法。

  1. @Override
  2. public JobManagerRunner createJobManagerRunner(
  3.         JobGraph jobGraph,
  4.         Configuration configuration,
  5.         RpcService rpcService,
  6.         HighAvailabilityServices highAvailabilityServices,
  7.         HeartbeatServices heartbeatServices,
  8.         JobManagerSharedServices jobManagerServices,
  9.         JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory,
  10.         FatalErrorHandler fatalErrorHandler,
  11.         long initializationTimestamp) throws Exception {
  13.     // // 创建JobManager的配置
  14.     final JobMasterConfiguration jobMasterConfiguration = JobMasterConfiguration.fromConfiguration(configuration);
  16.     // 创建SlotPool服务和Scheduler的工厂类
  17.     // 该工厂类用于创建SlotPoolService和SchedulerNG
  18.     final SlotPoolFactory slotPoolFactory = SlotPoolFactory.fromConfiguration(configuration);
  19.     final SchedulerNGFactory schedulerNGFactory = SchedulerNGFactoryFactory.createSchedulerNGFactory(configuration);
  20.     final ShuffleMaster<?> shuffleMaster = ShuffleServiceLoader.loadShuffleServiceFactory(configuration).createShuffleMaster(configuration);
  22.     // 创建JobMaster服务工厂
  23.     // JobMaster服务用于获取JobManaster所在地址，获取和JobMaster通信的Gateway
  24.     final JobMasterServiceFactory jobMasterFactory = new DefaultJobMasterServiceFactory(
  25.         jobMasterConfiguration,
  26.         slotPoolFactory,
  27.         rpcService,
  28.         highAvailabilityServices,
  29.         jobManagerServices,
  30.         heartbeatServices,
  31.         jobManagerJobMetricGroupFactory,
  32.         fatalErrorHandler,
  33.         schedulerNGFactory,
  34.         shuffleMaster);
  36.     // 创建JobManagerRunnerImpl
  37.     return new JobManagerRunnerImpl(
  38.         jobGraph,
  39.         jobMasterFactory,
  40.         highAvailabilityServices,
  41.         jobManagerServices.getLibraryCacheManager().registerClassLoaderLease(jobGraph.getJobID()),
  42.         jobManagerServices.getScheduledExecutorService(),
  43.         fatalErrorHandler,
  44.         initializationTimestamp);
  45. }

JobManagerRunner

JobManagerRunner用于启动JobManager。JobManagerRunner在创建时需要获取用户代码类加载器，RunningJobsRegistry（用于跟踪job执行状态，等待执行，执行中或者是执行完毕）和leader选举服务。

接下来需要启动JobManager。我们查看下start方法：

  1. @Override
  2. public void start() throws Exception {
  3.     try {
  4.         leaderElectionService.start(this);
  5.     } catch (Exception e) {
  6.         log.error("Could not start the JobManager because the leader election service did not start.", e);
  7.         throw new Exception("Could not start the leader election service.", e);
  8.     }
  9. }

start方法就一个任务：启动leader选举服务。启动leader选举过程。leader选举过程详细分析将在后续补充完整！

一旦有JobManagerRunner实例被授予leader角色，它的grantLeadership方法会被调用。grantLeadership是LeaderContender接口的方法，所有参与leader竞选的角色都必须要实现这个接口。JobManagerRunner自然也不例外。
我们查看grantLeadership方法：

  1. @Override
  2. public void grantLeadership(final UUID leaderSessionID) {
  3.     synchronized (lock) {
  4.         if (shutdown) {
  5.             log.debug("JobManagerRunner cannot be granted leadership because it is already shut down.");
  6.         return;
  7.         }
  9.         leadershipOperation = leadershipOperation.thenCompose(
  10.             (ignored) -> {
  11.                 synchronized (lock) {
  12.                     return verifyJobSchedulingStatusAndStartJobManager(leaderSessionID);
  13.                 }
  14.             });
  16.         handleException(leadershipOperation, "Could not start the job manager.");
  17.     }
  18. }

接下来到了校验作业调度和启动JobManager的时候。这部分逻辑位于verifyJobSchedulingStatusAndStartJobManager方法。

  1. private CompletableFuture<Void> verifyJobSchedulingStatusAndStartJobManager(UUID leaderSessionId) {
  2.     // 检查作业调度状态
  3.     final CompletableFuture<JobSchedulingStatus> jobSchedulingStatusFuture = getJobSchedulingStatus();
  5.     return jobSchedulingStatusFuture.thenCompose(
  6.         jobSchedulingStatus -> {
  7.             // 如果作业已经执行完毕，调用执行完毕逻辑
  8.             if (jobSchedulingStatus == JobSchedulingStatus.DONE) {
  9.                 return jobAlreadyDone();
  10.             } else {
  11.                 // 否则，启动JobMaster
  12.                 return startJobMaster(leaderSessionId);
  13.             }
  14.         });
  15. }

这么长一路下来，终于到了启动JobMaster的时候。

  1. private CompletionStage<Void> startJobMaster(UUID leaderSessionId) {
  2.     log.info("JobManager runner for job {} ({}) was granted leadership with session id {} at {}.",
  3.         jobGraph.getName(), jobGraph.getJobID(), leaderSessionId, jobMasterService.getAddress());
  5.     try {
  6.         // 先设置作业的状态为正在运行
  7.         runningJobsRegistry.setJobRunning(jobGraph.getJobID());
  8.     } catch (IOException e) {
  9.         return FutureUtils.completedExceptionally(
  10.             new FlinkException(
  11.                 String.format("Failed to set the job %s to running in the running jobs registry.", jobGraph.getJobID()),
  12.                 e));
  13.     }
  15.     final CompletableFuture<Acknowledge> startFuture;
  16.     try {
  17.         // 然后启动JobMaster服务
  18.         startFuture = jobMasterService.start(new JobMasterId(leaderSessionId));
  19.     } catch (Exception e) {
  20.         return FutureUtils.completedExceptionally(new FlinkException("Failed to start the JobMaster.", e));
  21.     }
  23.     final CompletableFuture<JobMasterGateway> currentLeaderGatewayFuture = leaderGatewayFuture;
  24.     return startFuture.thenAcceptAsync(
  25.         (Acknowledge ack) -> confirmLeaderSessionIdIfStillLeader(
  26.             leaderSessionId,
  27.             jobMasterService.getAddress(),
  28.             currentLeaderGatewayFuture),
  29.         executor);
  30. }

到这里为止，执行过程流转到了JobMaster对象。

JobMaster

JobMaster负责执行一个JobGraph。

  1. public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception {
  2.     // make sure we receive RPC and async calls
  3.     start();
  5.     return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT);
  6. }

start方法调用了RPC的start方法

startJobExecution方法启动JobMaster服务和开始任务调度：

  1. private Acknowledge startJobExecution(JobMasterId newJobMasterId) throws Exception {
  3.     // 检查是否在主线程执行
  4.     validateRunsInMainThread();
  6.     checkNotNull(newJobMasterId, "The new JobMasterId must not be null.");
  8.     if (Objects.equals(getFencingToken(), newJobMasterId)) {
  9.         log.info("Already started the job execution with JobMasterId {}.", newJobMasterId);
  11.         return Acknowledge.get();
  12.     }
  14.     setNewFencingToken(newJobMasterId);
  16.     startJobMasterServices();
  18.     log.info("Starting execution of job {} ({}) under job master id {}.", jobGraph.getName(), jobGraph.getJobID(), newJobMasterId);
  20.     resetAndStartScheduler();
  22.     return Acknowledge.get();
  23. }

startJobMasterServices方法创建出TaskManager心跳管理器，启动SlotPoolService和建立起ResourceManager leader的连接（ResourceManager也有leader选举过程）。

  1. private void startJobMasterServices() throws Exception {
  2.     startHeartbeatServices();
  4.     // start the slot pool make sure the slot pool now accepts messages for this leader
  5.     slotPool.start(getFencingToken(), getAddress(), getMainThreadExecutor());
  7.     //TODO: Remove once the ZooKeeperLeaderRetrieval returns the stored address upon start
  8.     // try to reconnect to previously known leader
  9.     reconnectToResourceManager(new FlinkException("Starting JobMaster component."));
  11.     // job is ready to go, try to establish connection with resource manager
  12.     //   - activate leader retrieval for the resource manager
  13.     //   - on notification of the leader, the connection will be established and
  14.     //     the slot pool will start requesting slots
  15.     resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
  16. }

我们回到上面的resetAndStartScheduler方法，查看它的代码

  1. private void resetAndStartScheduler() throws Exception {
  2.     validateRunsInMainThread();
  4.     final CompletableFuture<Void> schedulerAssignedFuture;
  6.     if (schedulerNG.requestJobStatus() == JobStatus.CREATED) {
  7.         schedulerAssignedFuture = CompletableFuture.completedFuture(null);
  8.         schedulerNG.setMainThreadExecutor(getMainThreadExecutor());
  9.     } else {
  10.         suspendAndClearSchedulerFields(new FlinkException("ExecutionGraph is being reset in order to be rescheduled."));
  11.         final JobManagerJobMetricGroup newJobManagerJobMetricGroup = jobMetricGroupFactory.create(jobGraph);
  12.         final SchedulerNG newScheduler = createScheduler(executionDeploymentTracker, newJobManagerJobMetricGroup);
  14.         schedulerAssignedFuture = schedulerNG.getTerminationFuture().handle(
  15.             (ignored, throwable) -> {
  16.                 newScheduler.setMainThreadExecutor(getMainThreadExecutor());
  17.                 assignScheduler(newScheduler, newJobManagerJobMetricGroup);
  18.                 return null;
  19.             }
  20.         );
  21.     }
  23.     FutureUtils.assertNoException(schedulerAssignedFuture.thenRun(this::startScheduling));
  24. }

我们回到上面的startScheduling方法，查看它的代码

  1. private void startScheduling() {
  2.     checkState(jobStatusListener == null);
  3.     // register self as job status change listener
  4.     jobStatusListener = new JobManagerJobStatusListener();
  5.     schedulerNG.registerJobStatusListener(jobStatusListener);
  7.     schedulerNG.startScheduling();
  8. }

SchedulerNG

SchedulerNG为Flink的调度器接口，负责根据JobGraph创建ExecutionGraph然后将作业调度执行。

下面的分析我们已默认的DefaultScheduler为准进行分析。DefaultScheduler的父类为SchedulerBase。它在初始化的时候将JobGraph转换为ExecutionGraph。

  1. @Override
  2. public final void startScheduling() {
  3.     mainThreadExecutor.assertRunningInMainThread();
  4.     registerJobMetrics();
  5.     startAllOperatorCoordinators();
  6.     startSchedulingInternal();
  7. }

我们回到开始调度执行的逻辑。SchedulerBase的startScheduling方法调用了startSchedulingInternal。

  1. @Override
  2. protected void startSchedulingInternal() {
  3.     log.info("Starting scheduling with scheduling strategy [{}]", schedulingStrategy.getClass().getName());
  4.     // 设置ExecutionGraph的JobStatus状态为Running
  5.     prepareExecutionGraphForNgScheduling();
  6.     // 执行调度策略的startScheduling方法
  7.     schedulingStrategy.startScheduling();
  8. }

接下来我们查看下SchedulingStrategy唯一的实现类PipelinedRegionSchedulingStrategy的startScheduling方法。

  1. @Override
  2. public void startScheduling() {
  3.     final Set<SchedulingPipelinedRegion> sourceRegions = IterableUtils
  4.         .toStream(schedulingTopology.getAllPipelinedRegions())
  5.         .filter(region -> !region.getConsumedResults().iterator().hasNext())
  6.         .collect(Collectors.toSet());
  7.     maybeScheduleRegions(sourceRegions);
  8. }

此方法先创建出sourceRegions集合。获取所有的Pipeline类型的Region。然后过滤掉其中不包含consumed result的region（region从上游region接收的数据称为consumed result，为下游region输出的数据称为produced result），即最后剩下的是pipelined region的源头节点。

接下来轮到maybeScheduleRegions方法。

  1. private void maybeScheduleRegions(final Set<SchedulingPipelinedRegion> regions) {
  2.     final List<SchedulingPipelinedRegion> regionsSorted =
  3.         SchedulingStrategyUtils.sortPipelinedRegionsInTopologicalOrder(schedulingTopology, regions);
  4.     for (SchedulingPipelinedRegion region : regionsSorted) {
  5.         maybeScheduleRegion(region);
  6.     }
  7. }

该方法将这些region安装拓扑顺序排序后，逐个调用maybeScheduleRegion方法。

  1. private void maybeScheduleRegion(final SchedulingPipelinedRegion region) {
  2.     // 如果region中有一个consumed result状态不是CONSUMABLE（数据可以被消费），返回
  3.     if (!areRegionInputsAllConsumable(region)) {
  4.         return;
  5.     }
  6.     // 检查region中所有的节点必须为已创建状态
  7.     checkState(areRegionVerticesAllInCreatedState(region), "BUG: trying to schedule a region which is not in CREATED state");
  8.     // 创建各个Execution节点和部署选项
  9.     final List<ExecutionVertexDeploymentOption> vertexDeploymentOptions =
  10.         SchedulingStrategyUtils.createExecutionVertexDeploymentOptions(
  11.             regionVerticesSorted.get(region),
  12.             id -> deploymentOption);
  13.     // 开始为执行节点分配资源（slot）和部署
  14.     schedulerOperations.allocateSlotsAndDeploy(vertexDeploymentOptions);
  15. }

SchedulerOperations接口是SchedulingStrategy用于实现调度决策的方法。它拥有一个实现类DefaultScheduler。

  1. @Override
  2. public void allocateSlotsAndDeploy(final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions) {
  3.     // 检查每个ExecutionVertex的状态必须为ExecutionState.CREATED
  4.     validateDeploymentOptions(executionVertexDeploymentOptions);
  6.     // 将ExecutionVertexID提取成key
  7.     final Map<ExecutionVertexID, ExecutionVertexDeploymentOption> deploymentOptionsByVertex =
  8.         groupDeploymentOptionsByVertexId(executionVertexDeploymentOptions);
  10.     // 提取出所有的ExecutionVertexID
  11.     final List<ExecutionVertexID> verticesToDeploy = executionVertexDeploymentOptions.stream()
  12.         .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
  13.         .collect(Collectors.toList());
  15.     // 创建每个ExecutionVertex的版本信息，默认值为1L
  16.     final Map<ExecutionVertexID, ExecutionVertexVersion> requiredVersionByVertex =
  17.         executionVertexVersioner.recordVertexModifications(verticesToDeploy);
  19.     // 将每个ExecutionVertex的状态切换为ExecutionState.SCHEDULED
  20.     transitionToScheduled(verticesToDeploy);
  22.     // 为所有vertex分配slot，即执行所需的资源
  23.     final List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments =
  24.         allocateSlots(executionVertexDeploymentOptions);
  25.     // 创建vertex部署相关信息
  26.     // 即ExecutionVertexVersion，ExecutionVertexDeploymentOption和SlotExecutionVertexAssignment的包装类
  27.     final List<DeploymentHandle> deploymentHandles = createDeploymentHandles(
  28.         requiredVersionByVertex,
  29.         deploymentOptionsByVertex,
  30.         slotExecutionVertexAssignments);
  31.     // 等待所有节点分配资源和部署完毕
  32.     waitForAllSlotsAndDeploy(deploymentHandles);
  33. }

我们继续跟踪waitForAllSlotsAndDeploy方法

  1. private void waitForAllSlotsAndDeploy(final List<DeploymentHandle> deploymentHandles) {
  2.     FutureUtils.assertNoException(
  3.         assignAllResources(deploymentHandles).handle(deployAll(deploymentHandles)));
  4. }

assignAllResources方法给所有的vertex分配执行所需的资源，然后执行deployAll部署所有节点。

  1. private BiFunction<Void, Throwable, Void> deployAll(final List<DeploymentHandle> deploymentHandles) {
  2.     return (ignored, throwable) -> {
  3.         propagateIfNonNull(throwable);
  4.         for (final DeploymentHandle deploymentHandle : deploymentHandles) {
  5.             final SlotExecutionVertexAssignment slotExecutionVertexAssignment = deploymentHandle.getSlotExecutionVertexAssignment();
  6.             final CompletableFuture<LogicalSlot> slotAssigned = slotExecutionVertexAssignment.getLogicalSlotFuture();
  7.             // 确保slot分配已经完成
  8.             checkState(slotAssigned.isDone());
  9.             // 在slot分配完成后执行deployOrHandleError方法
  10.             FutureUtils.assertNoException(
  11.                 slotAssigned.handle(deployOrHandleError(deploymentHandle)));
  12.         }
  13.         return null;
  14.     };
  15. }

接着查看deployOrHandleError方法，代码如下：

  1. private BiFunction<Object, Throwable, Void> deployOrHandleError(final DeploymentHandle deploymentHandle) {
  2.     final ExecutionVertexVersion requiredVertexVersion = deploymentHandle.getRequiredVertexVersion();
  3.     final ExecutionVertexID executionVertexId = requiredVertexVersion.getExecutionVertexId();
  5.     return (ignored, throwable) -> {
  6.         // 检查ExecutionVertex的版本是否已经修改
  7.         // 如果已经修改，说明该vertex被其他deployment部署，终止本次部署操作
  8.         if (executionVertexVersioner.isModified(requiredVertexVersion)) {
  9.             log.debug("Refusing to deploy execution vertex {} because this deployment was " +
  10.                 "superseded by another deployment", executionVertexId);
  11.             return null;
  12.         }
  14.         if (throwable == null) {
  15.             // 部署ExecutionVertex
  16.             deployTaskSafe(executionVertexId);
  17.         } else {
  18.             handleTaskDeploymentFailure(executionVertexId, throwable);
  19.         }
  20.         return null;
  21.     };
  22. }

接着跳转到deployTaskSafe方法：

  1. private void deployTaskSafe(final ExecutionVertexID executionVertexId) {
  2.     try {
  3.         final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);
  4.         // 在获取ExecutionVertex之后部署节点
  5.         executionVertexOperations.deploy(executionVertex);
  6.     } catch (Throwable e) {
  7.         handleTaskDeploymentFailure(executionVertexId, e);
  8.     }
  9. }

查看下ExecutionVertexOperations的唯一实现类DefaultExecutionVertexOperations的deploy方法：

  1. @Override
  2. public void deploy(final ExecutionVertex executionVertex) throws JobException {
  3.     executionVertex.deploy();
  4. }

该方法调用了ExecutionVertex的部署方法。

ExecutionVertex

ExecutionVertex的deploy方法内容如下。

  1. public void deploy() throws JobException {
  2.     currentExecution.deploy();
  3. }

其中currentExecution为Execution对象。每次尝试执行ExecutionVertex都会创建出一个Execution对象。currentExecution变量保存了最近一次创建的Execution。

Execution的deploy方法内容请见下面说明：

  1. /**
  2.  * Deploys the execution to the previously assigned resource.
  3.  *
  4.  * @throws JobException if the execution cannot be deployed to the assigned resource
  5.  */
  6. public void deploy() throws JobException {
  7.     assertRunningInJobMasterMainThread();
  9.     final LogicalSlot slot  = assignedResource;
  11.     checkNotNull(slot, "In order to deploy the execution we first have to assign a resource via tryAssignResource.");
  13.     // Check if the TaskManager died in the meantime
  14.     // This only speeds up the response to TaskManagers failing concurrently to deployments.
  15.     // The more general check is the rpcTimeout of the deployment call
  16.     // 资源必须是可用状态
  17.     if (!slot.isAlive()) {
  18.         throw new JobException("Target slot (TaskManager) for deployment is no longer alive.");
  19.     }
  21.     // make sure exactly one deployment call happens from the correct state
  22.     // note: the transition from CREATED to DEPLOYING is for testing purposes only
  23.     ExecutionState previous = this.state;
  24.     // 执行状态从SCHEDULED或CREATED转换成DEPLOYING
  25.     if (previous == SCHEDULED || previous == CREATED) {
  26.         if (!transitionState(previous, DEPLOYING)) {
  27.             // race condition, someone else beat us to the deploying call.
  28.             // this should actually not happen and indicates a race somewhere else
  29.             throw new IllegalStateException("Cannot deploy task: Concurrent deployment call race.");
  30.         }
  31.     }
  32.     else {
  33.         // vertex may have been cancelled, or it was already scheduled
  34.         throw new IllegalStateException("The vertex must be in CREATED or SCHEDULED state to be deployed. Found state " + previous);
  35.     }
  37.     // 检查slot是否分配给了当前这个Execution
  38.     if (this != slot.getPayload()) {
  39.         throw new IllegalStateException(
  40.             String.format("The execution %s has not been assigned to the assigned slot.", this));
  41.     }
  43.     try {
  45.         // race double check, did we fail/cancel and do we need to release the slot?
  46.         // 再次检查状态是否是部署中（DEPLOYING）
  47.         if (this.state != DEPLOYING) {
  48.             slot.releaseSlot(new FlinkException("Actual state of execution " + this + " (" + state + ") does not match expected state DEPLOYING."));
  49.             return;
  50.         }
  52.         LOG.info("Deploying {} (attempt #{}) with attempt id {} to {} with allocation id {}", vertex.getTaskNameWithSubtaskIndex(),
  53.             attemptNumber, vertex.getCurrentExecutionAttempt().getAttemptId(), getAssignedResourceLocation(), slot.getAllocationId());
  55.         if (taskRestore != null) {
  56.             checkState(taskRestore.getTaskStateSnapshot().getSubtaskStateMappings().stream().allMatch(entry ->
  57.                 entry.getValue().getInputRescalingDescriptor().equals(InflightDataRescalingDescriptor.NO_RESCALE) &&
  58.                 entry.getValue().getOutputRescalingDescriptor().equals(InflightDataRescalingDescriptor.NO_RESCALE)),
  59.                 "Rescaling from unaligned checkpoint is not yet supported.");
  60.         }
  62.         // 创建Task部署描述符，用来创建Task
  63.         final TaskDeploymentDescriptor deployment = TaskDeploymentDescriptorFactory
  64.             .fromExecutionVertex(vertex, attemptNumber)
  65.             .createDeploymentDescriptor(
  66.                 slot.getAllocationId(),
  67.                 slot.getPhysicalSlotNumber(),
  68.                 taskRestore,
  69.                 producedPartitions.values());
  71.         // null taskRestore to let it be GC'ed
  72.         taskRestore = null;
  73.         // 获取提供资源的TaskManagerGateway
  74.         // 用来和TaskManager通信
  75.         final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
  77.         final ComponentMainThreadExecutor jobMasterMainThreadExecutor =
  78.             vertex.getExecutionGraph().getJobMasterMainThreadExecutor();
  80.         getVertex().notifyPendingDeployment(this);
  81.         // We run the submission in the future executor so that the serialization of large TDDs does not block
  82.         // the main thread and sync back to the main thread once submission is completed.
  83.         // RPC调用，告诉TaskManager创建一个Task，执行当前Execution
  84.         CompletableFuture.supplyAsync(() -> taskManagerGateway.submitTask(deployment, rpcTimeout), executor)
  85.             .thenCompose(Function.identity())
  86.             .whenCompleteAsync(
  87.                 (ack, failure) -> {
  88.                     if (failure == null) {
  89.                         vertex.notifyCompletedDeployment(this);
  90.                     } else {
  91.                         if (failure instanceof TimeoutException) {
  92.                             String taskname = vertex.getTaskNameWithSubtaskIndex() + " (" + attemptId + ')';
  94.                             markFailed(new Exception(
  95.                                 "Cannot deploy task " + taskname + " - TaskManager (" + getAssignedResourceLocation()
  96.                                     + ") not responding after a rpcTimeout of " + rpcTimeout, failure));
  97.                         } else {
  98.                             markFailed(failure);
  99.                         }
  100.                     }
  101.                 },
  102.                 jobMasterMainThreadExecutor);
  104.     }
  105.     catch (Throwable t) {
  106.         markFailed(t);
  108.         if (isLegacyScheduling()) {
  109.             ExceptionUtils.rethrow(t);
  110.         }
  111.     }
  112. }

最后，我们经历了种种曲折又复杂的过程，终于到了TaskManager执行task这一步。Task的创建和执行作者打算单独开篇来讲解。到此为止，Flink作业执行流程已分析完毕。