flink学习创建yarn客户端应用程序

程序入口

CliFrontend.java

public static void main(final String[] args) {
    EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);

    // 1. find the configuration directory
    /*TODO 获取flink的conf目录的路径*/
    final String configurationDirectory = getConfigurationDirectoryFromEnv();

    // 2. load the global configuration
    /*TODO 根据conf路径,加载配置*/
    final Configuration configuration = GlobalConfiguration.loadConfiguration(configurationDirectory);

    // 3. load the custom command lines
    /*TODO 封装命令行接口:按顺序:Generic、Yarn、Default*/
    final List<CustomCommandLine> customCommandLines = loadCustomCommandLines(
      configuration,
      configurationDirectory);

    try {
      final CliFrontend cli = new CliFrontend(
        configuration,
        customCommandLines);

      SecurityUtils.install(new SecurityConfiguration(cli.configuration));
      int retCode = SecurityUtils.getInstalledContext()
          .runSecured(() -> cli.parseAndRun(args));
      System.exit(retCode);
    }
    catch (Throwable t) {
      final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
      LOG.error("Fatal error while running command line interface.", strippedThrowable);
      strippedThrowable.printStackTrace();
      System.exit(31);
    }
  }

public int parseAndRun(String[] args) {

    // check for action
    if (args.length < 1) {
      CliFrontendParser.printHelp(customCommandLines);
      System.out.println("Please specify an action.");
      return 1;
    }

    // get action
    String action = args[0];

    // remove action from parameters
    final String[] params = Arrays.copyOfRange(args, 1, args.length);

    try {
      // do action
      switch (action) {
        case ACTION_RUN:
          run(params);
          return 0;
        case ACTION_RUN_APPLICATION:
          runApplication(params);
          return 0;
        case ACTION_LIST:
          list(params);
          return 0;
        case ACTION_INFO:
          info(params);
          return 0;
        case ACTION_CANCEL:
          cancel(params);
          return 0;
        case ACTION_STOP:
          stop(params);
          return 0;
        case ACTION_SAVEPOINT:
          savepoint(params);
          return 0;
        case "-h":
        case "--help":
          CliFrontendParser.printHelp(customCommandLines);
          return 0;
        case "-v":
        case "--version":
          String version = EnvironmentInformation.getVersion();
          String commitID = EnvironmentInformation.getRevisionInformation().commitId;
          System.out.print("Version: " + version);
          System.out.println(commitID.equals(EnvironmentInformation.UNKNOWN) ? "" : ", Commit ID: " + commitID);
          return 0;
        default:
          System.out.printf("\"%s\" is not a valid action.\n", action);
          System.out.println();
          System.out.println("Valid actions are \"run\", \"list\", \"info\", \"savepoint\", \"stop\", or \"cancel\".");
          System.out.println();
          System.out.println("Specify the version option (-v or --version) to print Flink version.");
          System.out.println();
          System.out.println("Specify the help option (-h or --help) to get help on the command.");
          return 1;
      }
    } catch (CliArgsException ce) {
      return handleArgException(ce);
    } catch (ProgramParametrizationException ppe) {
      return handleParametrizationException(ppe);
    } catch (ProgramMissingJobException pmje) {
      return handleMissingJobException();
    } catch (Exception e) {
      return handleError(e);
    }
  }