1.flink.sh文件

Flink提交任务的脚本为flink.sh，该文件位于 flink-dist/src/main/flink-bin/bin/ 目录下，我们去寻找flink.sh中Java入口代码。

  1. ## 省略
  2. exec 
  3. $JAVA_RUN $JVM_ARGS 
  4. $FLINK_ENV_JAVA_OPTS 
  5. "${log_setting[@]}" 
  6. -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" 
  7. org.apache.flink.client.cli.CliFrontend "$@"

通过flink.sh不难发现Flink提交任务的入口类为CliFrontend。找到这个类的main方法：

  1. /**
  2. * Submits the job based on the arguments.
  3. */
  4. public static void main(final String[] args) {
  5.     // 日志打印环境配置信息
  6.     // 例如：代码版本，当前用户，Java版本，jvm参数
  7.     EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);
  9.     // 1. find the configuration directory
  10.     // 读取flink配置文件目录
  11.     final String configurationDirectory = getConfigurationDirectoryFromEnv();
  13.     // 2. load the global configuration
  14.     // 读取flink配置文件到configuration对象中
  15.     final Configuration configuration = GlobalConfiguration.loadConfiguration(configurationDirectory);
  17.     // 3. load the custom command lines
  18.     // 加载自定义命令行
  19.     final List<CustomCommandLine> customCommandLines = loadCustomCommandLines(
  20.         configuration,
  21.         configurationDirectory);
  23.     try {
  24.         // 实例化 CliFrontend
  25.         final CliFrontend cli = new CliFrontend(
  26.             configuration,
  27.             customCommandLines);
  29.         // 安装认证配置信息
  30.         SecurityUtils.install(new SecurityConfiguration(cli.configuration));
  31.         // 在配置环境下调用启动参数
  32.         int retCode = SecurityUtils.getInstalledContext()
  33.             .runSecured(() -> cli.parseAndRun(args));
  34.         System.exit(retCode);
  35.     }
  36.     catch (Throwable t) {
  37.         final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
  38.         LOG.error("Fatal error while running command line interface.", strippedThrowable);
  39.         strippedThrowable.printStackTrace();
  40.         System.exit(31);
  41.     }
  42.     }

有以上分析后，我们发现flink的主要实现逻辑位于

  1. cli.parseAndRun(args)

在分析主要逻辑之前，我们注重分析下main方法的各个执行环节。

�getConfigurationDirectoryFromEnv方法的源码：

  1.     public static String getConfigurationDirectoryFromEnv() {
  2.         // 从FLINK_CONF_DIR环境变量获取Flink配置文件路径
  3.         String location = System.getenv(ConfigConstants.ENV_FLINK_CONF_DIR);
  4.         // 如果配置了该环境变量，并且目录存在，则直接返回
  5.         if (location != null) {
  6.             if (new File(location).exists()) {
  7.                 return location;
  8.             }
  9.             else {
  10.                 throw new RuntimeException("The configuration directory '" + location + "', specified in the '" +
  11.                     ConfigConstants.ENV_FLINK_CONF_DIR + "' environment variable, does not exist.");
  12.             }
  13.         }
  14.         // 如果没有配置环境变量，从上层目录的conf寻找（../conf）
  15.         else if (new File(CONFIG_DIRECTORY_FALLBACK_1).exists()) {
  16.             location = CONFIG_DIRECTORY_FALLBACK_1;
  17.         }
  18.         // 如果没有配置环境变量，从当前目录的conf寻找（conf）
  19.         else if (new File(CONFIG_DIRECTORY_FALLBACK_2).exists()) {
  20.             location = CONFIG_DIRECTORY_FALLBACK_2;
  21.         }
  22.         // 如果都没有找到，系统异常退出
  23.         else {
  24.             throw new RuntimeException("The configuration directory was not specified. " +
  25.                     "Please specify the directory containing the configuration file through the '" +
  26.                 ConfigConstants.ENV_FLINK_CONF_DIR + "' environment variable.");
  27.         }
  28.         return location;
  29.     }

loadCustomCommandLines方法源码：

  1.     public static List<CustomCommandLine> loadCustomCommandLines(Configuration configuration, String configurationDirectory) {
  2.         // GenericCLI --> flinkYarnSessionCLI --> DefaultCLI顺序加载命令行
  3.         List<CustomCommandLine> customCommandLines = new ArrayList<>();
  4.         customCommandLines.add(new GenericCLI(configuration, configurationDirectory));
  6.         //    Command line interface of the YARN session, with a special initialization here
  7.         //    to prefix all options with y/yarn.
  8.         // yarn回话的命令行接口，这里所有选项使用y/yarn前缀指定
  9.         final String flinkYarnSessionCLI = "org.apache.flink.yarn.cli.FlinkYarnSessionCli";
  10.         try {
  11.             customCommandLines.add(
  12.                 loadCustomCommandLine(flinkYarnSessionCLI,
  13.                     configuration,
  14.                     configurationDirectory,
  15.                     "y",
  16.                     "yarn"));
  17.         } catch (NoClassDefFoundError | Exception e) {
  18.             final String errorYarnSessionCLI = "org.apache.flink.yarn.cli.FallbackYarnSessionCli";
  19.             try {
  20.                 LOG.info("Loading FallbackYarnSessionCli");
  21.                 customCommandLines.add(
  22.                         loadCustomCommandLine(errorYarnSessionCLI, configuration));
  23.             } catch (Exception exception) {
  24.                 LOG.warn("Could not load CLI class {}.", flinkYarnSessionCLI, e);
  25.             }
  26.         }
  28.         //    Tips: DefaultCLI must be added at last, because getActiveCustomCommandLine(..) will get the
  29.         //          active CustomCommandLine in order and DefaultCLI isActive always return true.
  30.         // 注意：默认客户端必须最后添加，getActiveCustomCommandLine方法将会按照CustomCommandLine顺序判断
  31.         // DefaultCLI的isActive方法一直返回true
  32.         customCommandLines.add(new DefaultCLI());
  34.         return customCommandLines;
  35.     }

�接下来，分析主要逻辑 cli.parseAndRun(args) 方法

  1. /**
  2.      * Parses the command line arguments and starts the requested action.
  3.      * 解析命令行参数，并开始请求操作
  4.      * @param args command line arguments of the client.
  5.      * @return The return code of the program
  6.      */
  7.     public int parseAndRun(String[] args) {
  9.         // check for action
  10.         // 检查命令行 个数
  11.         if (args.length < 1) {
  12.             CliFrontendParser.printHelp(customCommandLines);
  13.             System.out.println("Please specify an action.");
  14.             return 1;
  15.         }
  17.         // get action
  18.         // 获取flink将要执行的操作
  19.         String action = args[0];
  21.         // remove action from parameters
  22.         // 移除操作参数
  23.         final String[] params = Arrays.copyOfRange(args, 1, args.length);
  25.         try {
  26.             // do action 行为操作主要有一下几种：
  27.             // run，run-application，info，list，cancel，stop，savepoint
  28.             switch (action) {
  29.                 case ACTION_RUN:
  30.                     run(params);
  31.                     return 0;
  32.                 case ACTION_RUN_APPLICATION:
  33.                     runApplication(params);
  34.                     return 0;
  35.                 case ACTION_LIST:
  36.                     list(params);
  37.                     return 0;
  38.                 case ACTION_INFO:
  39.                     info(params);
  40.                     return 0;
  41.                 case ACTION_CANCEL:
  42.                     cancel(params);
  43.                     return 0;
  44.                 case ACTION_STOP:
  45.                     stop(params);
  46.                     return 0;
  47.                 case ACTION_SAVEPOINT:
  48.                     savepoint(params);
  49.                     return 0;
  50.                 case "-h":
  51.                 case "--help":
  52.                     CliFrontendParser.printHelp(customCommandLines);
  53.                     return 0;
  54.                 case "-v":
  55.                 case "--version":
  56.                     String version = EnvironmentInformation.getVersion();
  57.                     String commitID = EnvironmentInformation.getRevisionInformation().commitId;
  58.                     System.out.print("Version: " + version);
  59.                     System.out.println(commitID.equals(EnvironmentInformation.UNKNOWN) ? "" : ", Commit ID: " + commitID);
  60.                     return 0;
  61.                 default:
  62.                     System.out.printf("\"%s\" is not a valid action.\n", action);
  63.                     System.out.println();
  64.                     System.out.println("Valid actions are \"run\", \"list\", \"info\", \"savepoint\", \"stop\", or \"cancel\".");
  65.                     System.out.println();
  66.                     System.out.println("Specify the version option (-v or --version) to print Flink version.");
  67.                     System.out.println();
  68.                     System.out.println("Specify the help option (-h or --help) to get help on the command.");
  69.                     return 1;
  70.             }
  71.         } catch (CliArgsException ce) {
  72.             return handleArgException(ce);
  73.         } catch (ProgramParametrizationException ppe) {
  74.             return handleParametrizationException(ppe);
  75.         } catch (ProgramMissingJobException pmje) {
  76.             return handleMissingJobException();
  77.         } catch (Exception e) {
  78.             return handleError(e);
  79.         }
  80.     }

接下来我们进一步分析run()方法

  1. /**
  2.      * Executions the run action.
  3.      *
  4.      * @param args Command line arguments for the run action.
  5.      */
  6.     protected void run(String[] args) throws Exception {
  7.         LOG.info("Running 'run' command.");
  9.         // 解析命令行参数
  10.         final Options commandOptions = CliFrontendParser.getRunCommandOptions();
  11.         final CommandLine commandLine = getCommandLine(commandOptions, args, true);
  13.         // evaluate help flag
  14.         if (commandLine.hasOption(HELP_OPTION.getOpt())) {
  15.             CliFrontendParser.printHelpForRun(customCommandLines);
  16.             return;
  17.         }
  19.         // 获取有效的命令行，上述加载的三种客户端形式
  20.         final CustomCommandLine activeCommandLine =
  21.                 validateAndGetActiveCommandLine(checkNotNull(commandLine));
  23.         final ProgramOptions programOptions = ProgramOptions.create(commandLine);
  25.         // jar包，并获取对应的依赖
  26.         final List<URL> jobJars = getJobJarAndDependencies(programOptions);
  28.         //
  29.         final Configuration effectiveConfiguration = getEffectiveConfiguration(
  30.                 activeCommandLine, commandLine, programOptions, jobJars);
  32.         LOG.debug("Effective executor configuration: {}", effectiveConfiguration);
  33.         // 程序包
  34.         final PackagedProgram program = getPackagedProgram(programOptions, effectiveConfiguration);
  36.         try {
  37.             executeProgram(effectiveConfiguration, program);
  38.         } finally {
  39.             program.deleteExtractedLibraries();
  40.         }
  41.     }

这段代码主要有三部分，解析运行参数，从用户提交的jar包封装为PackagedProgram，还有执行程序以及清理操作。

这里需要调研下如何获取活跃的commandLine

  1.     /**
  2.      * Gets the custom command-line for the arguments.
  3.      * @param commandLine The input to the command-line.
  4.      * @return custom command-line which is active (may only be one at a time)
  5.      */
  6.     public CustomCommandLine validateAndGetActiveCommandLine(CommandLine commandLine) {
  7.         LOG.debug("Custom commandlines: {}", customCommandLines);
  8.         for (CustomCommandLine cli : customCommandLines) {
  9.             LOG.debug("Checking custom commandline {}, isActive: {}", cli, cli.isActive(commandLine));
  10.             if (cli.isActive(commandLine)) {
  11.                 return cli;
  12.             }
  13.         }
  14.         throw new IllegalStateException("No valid command-line found.");
  15.     }

这个方法很简单，从CustomCommandLine中获取到一个active状态的commandLine会立即返回。CustomCommandLine在初始化CliFrontend时会先初始化FlinkYarnSessionCli，然后初始化DefaultCli。因此该方法会在FlinkYarnSessionCli为active时优先返回FlinkYarnSessionCli。对于DefaultCli，它的isActive方法总是返回true。
FlinkYarnSessionCli的isActive方法：

  1.     @Override
  2.     public boolean isActive(CommandLine commandLine) {
  3.         if (!super.isActive(commandLine)) {
  4.             return (isYarnPropertiesFileMode(commandLine) && yarnApplicationIdFromYarnProperties != null);
  5.         }
  6.         return true;
  7.     }

下面分析下executeProgram方法, 进一步调用ClientUtils.executeProgram方法。

  1. public static void executeProgram(
  2.             PipelineExecutorServiceLoader executorServiceLoader,
  3.             Configuration configuration,
  4.             PackagedProgram program,
  5.             boolean enforceSingleJobExecution,
  6.             boolean suppressSysout) throws ProgramInvocationException {
  7.         checkNotNull(executorServiceLoader);
  8.         // 用户代码类加载器
  9.         final ClassLoader userCodeClassLoader = program.getUserCodeClassLoader();
  10.         // 上下文类加载器
  11.         final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
  13.         try {
  14.             // 使用用户类加载器
  15.             Thread.currentThread().setContextClassLoader(userCodeClassLoader);
  17.             LOG.info("Starting program (detached: {})", !configuration.getBoolean(DeploymentOptions.ATTACHED));
  19.             //  Execution Environment for remote execution with the Client
  20.             // 执行环境与客户端远程执行
  21.             ContextEnvironment.setAsContext(
  22.                 executorServiceLoader,
  23.                 configuration,
  24.                 userCodeClassLoader,
  25.                 enforceSingleJobExecution,
  26.                 suppressSysout);
  27.             // 流上下文环境
  28.             StreamContextEnvironment.setAsContext(
  29.                 executorServiceLoader,
  30.                 configuration,
  31.                 userCodeClassLoader,
  32.                 enforceSingleJobExecution,
  33.                 suppressSysout);
  35.             try {
  36.                 // 这种方法假定上下文环境准备,或者执行将默认本地执行。
  37.                 program.invokeInteractiveModeForExecution();
  38.             } finally {
  39.                 ContextEnvironment.unsetAsContext();
  40.                 StreamContextEnvironment.unsetAsContext();
  41.             }
  42.         } finally {
  43.             Thread.currentThread().setContextClassLoader(contextClassLoader);
  44.         }
  45.     }

executeProgram方法主要完成加载器类替换用户类加载器，设置环境，执行程序；
invokeInteractiveModeForExecution方法主要逻辑如下所示：

  1.     /**
  2.      * This method assumes that the context environment is prepared, or the execution
  3.      * will be a local execution by default.
  4.      */
  5.     public void invokeInteractiveModeForExecution() throws ProgramInvocationException {
  6.         callMainMethod(mainClass, args);
  7.     }

callMainMethod开始调用主函数，callMainMethod的主要逻辑如下所示

  1. private static void callMainMethod(Class<?> entryClass, String[] args) throws ProgramInvocationException {
  2.         Method mainMethod;
  3.         // 类必须是public
  4.         if (!Modifier.isPublic(entryClass.getModifiers())) {
  5.             throw new ProgramInvocationException("The class " + entryClass.getName() + " must be public.");
  6.         }
  8.         try {
  9.             // 寻找main方法
  10.             mainMethod = entryClass.getMethod("main", String[].class);
  11.         } catch (NoSuchMethodException e) {
  12.             throw new ProgramInvocationException("The class " + entryClass.getName() + " has no main(String[]) method.");
  13.         } catch (Throwable t) {
  14.             throw new ProgramInvocationException("Could not look up the main(String[]) method from the class " +
  15.                 entryClass.getName() + ": " + t.getMessage(), t);
  16.         }
  18.         // main必须是static
  19.         if (!Modifier.isStatic(mainMethod.getModifiers())) {
  20.             throw new ProgramInvocationException("The class " + entryClass.getName() + " declares a non-static main method.");
  21.         }
  22.         // main必须是public
  23.         if (!Modifier.isPublic(mainMethod.getModifiers())) {
  24.             throw new ProgramInvocationException("The class " + entryClass.getName() + " declares a non-public main method.");
  25.         }
  27.         try {
  28.             // 开始调用main方法
  29.             mainMethod.invoke(null, (Object) args);
  30.         } catch (IllegalArgumentException e) {
  31.             throw new ProgramInvocationException("Could not invoke the main method, arguments are not matching.", e);
  32.         } catch (IllegalAccessException e) {
  33.             throw new ProgramInvocationException("Access to the main method was denied: " + e.getMessage(), e);
  34.         } catch (InvocationTargetException e) {
  35.             Throwable exceptionInMethod = e.getTargetException();
  36.             if (exceptionInMethod instanceof Error) {
  37.                 throw (Error) exceptionInMethod;
  38.             } else if (exceptionInMethod instanceof ProgramParametrizationException) {
  39.                 throw (ProgramParametrizationException) exceptionInMethod;
  40.             } else if (exceptionInMethod instanceof ProgramInvocationException) {
  41.                 throw (ProgramInvocationException) exceptionInMethod;
  42.             } else {
  43.                 throw new ProgramInvocationException("The main method caused an error: " + exceptionInMethod.getMessage(), exceptionInMethod);
  44.             }
  45.         } catch (Throwable t) {
  46.             throw new ProgramInvocationException("An error occurred while invoking the program's main method: " + t.getMessage(), t);
  47.         }
  48.     }

逻辑：
1.首先判断类是public
2.寻找到main方法
3.判断main是static
4.判断main是public
5.执行我们编写的业务代码

到这里便开始执行用户编写的代码