入口：

Clifronted执行用户代码入口类Main方法之后，执行env.execute()

  1.  public JobExecutionResult execute(String jobName) throws Exception {
  2.         Preconditions.checkNotNull(jobName, "Streaming Job name should not be null.");
  4.         // 获取 StreamGraph 并继续执行
  5.         return execute(getStreamGraph(jobName));
  6.     }

StreamGraph的生成

SPI创建ExecutorFactory

往下获取executor进行execute 提交任务

  1.  public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
  2.         checkNotNull(streamGraph, "StreamGraph cannot be null.");
  3.         checkNotNull(
  4.                 configuration.get(DeploymentOptions.TARGET),
  5.                 "No execution.target specified in your configuration file.");
  7.         //SPI获取到Pipeline的执行器工厂
  8.         final PipelineExecutorFactory executorFactory =
  9.                 executorServiceLoader.getExecutorFactory(configuration);
  11.         //加载进行提交
  12.         CompletableFuture<JobClient> jobClientFuture =
  13.                 executorFactory
  14.                         .getExecutor(configuration)
  15.                         .execute(streamGraph, configuration, userClassloader);
  17.       。。。。。
  18.     }

SPI加载的Executor
flink-clients模块SPI
image.png
获取到Executor之后，RemoteExecutorFactory会创建RemoteExecutor进入其executor方法
image.png

standLone/Yarn-Session进入此方法org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor#execute

  1.  public CompletableFuture<JobClient> execute(
  2.             @Nonnull final Pipeline pipeline,
  3.             @Nonnull final Configuration configuration,
  4.             @Nonnull final ClassLoader userCodeClassloader)
  5.             throws Exception {
  6.         final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
  8.         try (final ClusterDescriptor<ClusterID> clusterDescriptor =
  9.                 clusterClientFactory.createClusterDescriptor(configuration)) {
  10.             final ClusterID clusterID = clusterClientFactory.getClusterId(configuration);
  11.             checkState(clusterID != null);
  13.             final ClusterClientProvider<ClusterID> clusterClientProvider =
  14.                     clusterDescriptor.retrieve(clusterID);
  15.             ClusterClient<ClusterID> clusterClient = clusterClientProvider.getClusterClient();
  16.             return clusterClient
  17.                     .submitJob(jobGraph)
  18.                      }
  19.     }

AbstractSessionClusterExecutor-StandLone/yarnSession入口

  • 创建jobGraph
  • 获取集群描述符ClusterDescriptor
  • 获取client

  • submitJob提交任务

  • standLone获取RestClusterClient

  • yarnSession获取

集群描述符ClusterDescriptor

standLone获取的Descriptor为StandLoneClusterDescriptor
Yarn获取的Descriptor为YarnClusterDescriptor

获取Client

  1. clusterDescriptor.retrieve(clusterID).getClusterClient()

由ClusterDescriptor获取对应的Client，
standLone获取的Client为RestClusterClient
Yarn获取的Client为YarnClient

AbstractJobClusterExecutor-PerJob入口

  1.   @Override
  2.     public CompletableFuture<JobClient> execute(
  3.             @Nonnull final Pipeline pipeline,
  4.             @Nonnull final Configuration configuration,
  5.             @Nonnull final ClassLoader userCodeClassloader)
  6.             throws Exception {
  7.         // streamGraph 转换为 jobGraph
  8.         final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
  10.        // 获取集群描述器  创建并启动了yarn的客户端，包含了一些yarn flink的配置和环境信息
  11.         try (final ClusterDescriptor<ClusterID> clusterDescriptor =
  12.                 clusterClientFactory.createClusterDescriptor(configuration)) {
  14.             final ExecutionConfigAccessor configAccessor =
  15.                     ExecutionConfigAccessor.fromConfiguration(configuration);
  17.             //获取集群特有信息，资源配置 jobManager内存 taskManager内存 每个TM的slot数
  18.             final ClusterSpecification clusterSpecification =
  19.                     clusterClientFactory.getClusterSpecification(configuration);
  21.             final ClusterClientProvider<ClusterID> clusterClientProvider =
  22.                     //部署job集群
  23.                     clusterDescriptor.deployJobCluster(
  24.                             clusterSpecification, jobGraph, configAccessor.getDetachedMode());
  26.             return CompletableFuture.completedFuture(
  27.                     new ClusterClientJobClientAdapter<>(
  28.                             clusterClientProvider, jobGraph.getJobID(), userCodeClassloader));
  29.         }
  30.     }