Flink1.10任务提交流程分析(二)

public class YarnJobClusterExecutor extends AbstractJobClusterExecutor<ApplicationId, YarnClusterClientFactory> {
   public static final String NAME = "yarn-per-job";
   public YarnJobClusterExecutor() {
      super(new YarnClusterClientFactory());
   }
}

public class AbstractJobClusterExecutor<ClusterID, ClientFactory extends ClusterClientFactory<ClusterID>> implements PipelineExecutor {

   private static final Logger LOG = LoggerFactory.getLogger(AbstractJobClusterExecutor.class);
   //代表的就是YarnClusterClientFactory
   private final ClientFactory clusterClientFactory;

   public AbstractJobClusterExecutor(@Nonnull final ClientFactory clusterClientFactory) {
      this.clusterClientFactory = checkNotNull(clusterClientFactory);
   }

   //执行任务提交
   //pipeline 代表StreamGraph
   public CompletableFuture<JobClient> execute(@Nonnull final Pipeline pipeline, @Nonnull final Configuration configuration) throws Exception {
      //将StreamGraph转换为JobGraph
      final JobGraph jobGraph = ExecutorUtils.getJobGraph(pipeline, configuration);
     //创建提交任务的一些信息：YarnClusterDescriptor
      try (final ClusterDescriptor<ClusterID> clusterDescriptor = clusterClientFactory.createClusterDescriptor(configuration)) {
        //将配置信息封装在ExecutionConfigAccessor中
         final ExecutionConfigAccessor configAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
         //包含了提交任务所需资源描述：内存大小、并行度 
         final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);
         //提交任务
         final ClusterClientProvider<ClusterID> clusterClientProvider = clusterDescriptor
               .deployJobCluster(clusterSpecification, jobGraph, 
                                 //是否采用分离模式
                                 configAccessor.getDetachedMode());
         LOG.info("Job has been submitted with JobID " + jobGraph.getJobID());

         return CompletableFuture.completedFuture(
               new ClusterClientJobClientAdapter<>(clusterClientProvider, jobGraph.getJobID()));
      }
   }
}

private ClusterClientProvider<ApplicationId> deployInternal(
      ClusterSpecification clusterSpecification,
      String applicationName,
      String yarnClusterEntrypoint,
      @Nullable JobGraph jobGraph,
      boolean detached) throws Exception {
    //..... 会做一些检查工作: yarn队列是否存在、配置检查
    //校验资源大小等等
   ApplicationReport report = startAppMaster(
         flinkConfiguration,
         applicationName,
         yarnClusterEntrypoint,
         jobGraph,
         yarnClient,
         yarnApplication,
         validClusterSpecification);

   //....
}

private ApplicationReport startAppMaster(
      Configuration configuration,
      String applicationName,
      String yarnClusterEntrypoint,
      JobGraph jobGraph,
      YarnClient yarnClient,
      YarnClientApplication yarnApplication,
      ClusterSpecification clusterSpecification) throws Exception {

   // ------------------ Initialize the file systems -------------------------

   org.apache.flink.core.fs.FileSystem.initialize(
         configuration,
         PluginUtils.createPluginManagerFromRootFolder(configuration));

   //获取homeDir， 表示jar、log配置上传的路径， 一般表示在hdfs上
   //其路径为/user/hadoop, （hadoop表示的当前的用户）
   final FileSystem fs = FileSystem.get(yarnConfiguration);
   final Path homeDir = fs.getHomeDirectory();
   //提交到yarn的描述信息
   ApplicationSubmissionContext appContext = yarnApplication.getApplicationSubmissionContext();
   // 会被上传到hdfs的文件 并且被添加到classpath中
   Set<File> systemShipFiles = new HashSet<>(shipFiles.size());
   // 仅仅是会被上传到hdfs , 但是不会被添加到classpath
   Set<File> shipOnlyFiles = new HashSet<>();
   for (File file : shipFiles) {
      systemShipFiles.add(file.getAbsoluteFile());
   }

   final String logConfigFilePath = configuration.getString(YarnConfigOptionsInternal.APPLICATION_LOG_CONFIG_FILE);
   if (logConfigFilePath != null) {
      systemShipFiles.add(new File(logConfigFilePath));
   }
   //将flink_home/lib 下的文件添加到systemShipFiles、通过-yt指定的文件也在里面
   addLibFoldersToShipFiles(systemShipFiles);

   //将flink_home/plugins 下的文件添加到shipOnlyFiles
   addPluginsFoldersToShipFiles(shipOnlyFiles);

   final ApplicationId appId = appContext.getApplicationId();

   // zk-ha相关的配置
   String zkNamespace = getZookeeperNamespace();
   // no user specified cli argument for namespace?
   if (zkNamespace == null || zkNamespace.isEmpty()) {
      // namespace defined in config? else use applicationId as default.
      zkNamespace = configuration.getString(HighAvailabilityOptions.HA_CLUSTER_ID, String.valueOf(appId));
      setZookeeperNamespace(zkNamespace);
   }

   configuration.setString(HighAvailabilityOptions.HA_CLUSTER_ID, zkNamespace);

   if (HighAvailabilityMode.isHighAvailabilityModeActivated(configuration)) {
      // activate re-execution of failed applications
      appContext.setMaxAppAttempts(
            configuration.getInteger(
                  YarnConfigOptions.APPLICATION_ATTEMPTS.key(),
                  YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS));

      activateHighAvailabilitySupport(appContext);
   } else {
      // set number of application retries to 1 in the default case
      appContext.setMaxAppAttempts(
            configuration.getInteger(
                  YarnConfigOptions.APPLICATION_ATTEMPTS.key(),
                  1));
   }

  //userJarFiles  表示用户jar
   final Set<File> userJarFiles = (jobGraph == null)
         // not per-job submission
         ? Collections.emptySet()
         // add user code jars from the provided JobGraph
         : jobGraph.getUserJars().stream().map(f -> f.toUri()).map(File::new).collect(Collectors.toSet());

   //需要cache文件上传到hdfs，一般使用在文件共享中
   if (jobGraph != null) {
      for (Map.Entry<String, DistributedCache.DistributedCacheEntry> entry : jobGraph.getUserArtifacts().entrySet()) {
         org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(entry.getValue().filePath);
         // only upload local files
         if (!path.getFileSystem().isDistributedFS()) {
            Path localPath = new Path(path.getPath());
            Tuple2<Path, Long> remoteFileInfo =
               Utils.uploadLocalFileToRemote(fs, appId.toString(), localPath, homeDir, entry.getKey());
            jobGraph.setUserArtifactRemotePath(entry.getKey(), remoteFileInfo.f0.toString());
         }
      }

      jobGraph.writeUserArtifactEntriesToConfiguration();
   }

   //表示启动appMaster需要的资源文件，会从hdfs上下载
   final Map<String, LocalResource> localResources = new HashMap<>(2 + systemShipFiles.size() + userJarFiles.size());
   // 访问hdfs的安全设置
   final List<Path> paths = new ArrayList<>(2 + systemShipFiles.size() + userJarFiles.size());
   // 启动taskExecutor需要的资源文件
   StringBuilder envShipFileList = new StringBuilder();

   //几个uploadAndRegisterFiles  方法，将systemShipFiles、shipOnlyFiles、用户jar上传到hdfs

   if (userJarInclusion == YarnConfigOptions.UserJarInclusion.ORDER) {
      systemClassPaths.addAll(userClassPaths);
   }

   // normalize classpath by sorting
   Collections.sort(systemClassPaths); //系统的一些classpath 排序
   Collections.sort(userClassPaths); //用户classpath 排序

   // classPathBuilder: 存放classpath的信息
   StringBuilder classPathBuilder = new StringBuilder();
     /*
      * 构建classpath: shipFile-jar、user-jar、log4j、yaml配置文件
      */

   final Path yarnFilesDir = getYarnFilesDir(appId);
   FsPermission permission = new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE);
   fs.setPermission(yarnFilesDir, permission); // set permission for path.

   /*
    *中间一堆与安全相关的配置
    */

  //执行的java命令信息，启动YarnJobClusterEntrypoint 
   final ContainerLaunchContext amContainer = setupApplicationMasterContainer(
         yarnClusterEntrypoint,
         hasLogback,
         hasLog4j,
         hasKrb5,
         clusterSpecification.getMasterMemoryMB());

   if (UserGroupInformation.isSecurityEnabled()) {
      // set HDFS delegation tokens when security is enabled
      LOG.info("Adding delegation token to the AM container.");
      Utils.setTokensFor(amContainer, paths, yarnConfiguration);
   }

   amContainer.setLocalResources(localResources);
   fs.close();

   // Setup CLASSPATH and environment variables for ApplicationMaster
   final Map<String, String> appMasterEnv = new HashMap<>();
   /**
    * 配置环境变量参数  到  appMasterEnv中，在启动启动YarnJobClusterEntrypoint时用到，
    * 例如： classpath、hadoopUser、appId等
    */

   amContainer.setEnvironment(appMasterEnv);

    // 还有一堆设置提交任务队列、yarn任务名称的配置信息

   // add a hook to clean up in case deployment fails
   Thread deploymentFailureHook = new DeploymentFailureHook(yarnApplication, yarnFilesDir);
   Runtime.getRuntime().addShutdownHook(deploymentFailureHook);
   LOG.info("Submitting application master " + appId);
   //提交任务  
   yarnClient.submitApplication(appContext);

   /**
    *  获取任务状态
    */
}