• Flink on yarn 实战和源码分析

    版本：1.13.6

    目录

    Flink on yarn 的3种模式的使用

    yarn session 模式源码分析

    yarn per-job模式源码分析

    application模式源码分析

    Flink on yarn 的3种模式的使用

    Application Mode #

    ./bin/flink run-application -t yarn-application ./examples/streaming/TopSpeedWindowing.jar

    Per-Job Mode #

    ./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar

    Session Mode #

    ./bin/flink run -t yarn-session ./examples/streaming/TopSpeedWindowing.jar

    注意：application 方式 使用的action为 run-application，而 per-job session mode 都是 run -t。 为什么不统一呢，官方没注意到？

    yarn session 模式源码分析

    以样例TopSpeedWindowing.jar为例，命令行提交命令为：

    ./bin/flink run -t yarn-session ./examples/streaming/TopSpeedWindowing.jar

    根据bin/flink脚本中的入口类 org.apache.flink.client.cli.CliFrontend

    1、找到bin/flink 提交脚本中看到启动类即程序的入口是： org.apache.flink.client.cli.CliFrontend

    2、查看其中的main方法，执行的逻辑简单总结如下：

    1. 获取flink的conf目录的路径

    2. 根据conf路径，加载配置

    3. 封装命令行接口：按顺序Generic、Yarn、Default   

    1. /** Submits the job based on the arguments. */
    2. public static void main(final String[] args) {
    3.     EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);
    4.  
    5.     // 1. find the configuration directory
    6.     final String configurationDirectory = getConfigurationDirectoryFromEnv();
    7.  
    8.     // 2. load the global configuration
    9.     final Configuration configuration =
    10.             GlobalConfiguration.loadConfiguration(configurationDirectory);
    11.  
    12.     // 3. load the custom command lines
    13.     final List customCommandLines =
    14.             loadCustomCommandLines(configuration, configurationDirectory);
    15.  
    16.     int retCode = 31;
    17.     try {
    18.         final CliFrontend cli = new CliFrontend(configuration, customCommandLines);
    19.  
    20.         SecurityUtils.install(new SecurityConfiguration(cli.configuration));
    21.         retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));
    22.     } catch (Throwable t) {
    23.         final Throwable strippedThrowable =
    24.                 ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
    25.         LOG.error("Fatal error while running command line interface.", strippedThrowable);
    26.         strippedThrowable.printStackTrace();
    27.     } finally {
    28.         System.exit(retCode);
    29.     }
    30. }

    parseAndRun 方法中 根据命令行的第一个参数匹配action ,此模式参数为run ,所以 case ACTION_RUN:

    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) {
    8.  
    9.     // check for action
    10.     if (args.length < 1) {
    11.         CliFrontendParser.printHelp(customCommandLines);
    12.         System.out.println("Please specify an action.");
    13.         return 1;
    14.     }
    15.  
    16.     // get action
    17.     String action = args[0];
    18.  
    19.     // remove action from parameters
    20.     final String[] params = Arrays.copyOfRange(args, 1, args.length);
    21.  
    22.     try {
    23.         // do action
    24.         switch (action) {
    25.             case ACTION_RUN:
    26.                 run(params);
    27.                 return 0;
    28.             case ACTION_RUN_APPLICATION:
    29.                 runApplication(params);
    30.                 return 0;
    31.             case ACTION_LIST:
    32.                 list(params);
    33.                 return 0;
    34.             case ACTION_INFO:
    35.                 info(params);
    36.                 return 0;
    37.             case ACTION_CANCEL:
    38.                 cancel(params);
    39.                 return 0;
    40.             case ACTION_STOP:
    41.                 stop(params);
    42.                 return 0;
    43.             case ACTION_SAVEPOINT:
    44.                 savepoint(params);
    45.                 return 0;
    46.             case "-h":
    47.             case "--help":
    48.                 CliFrontendParser.printHelp(customCommandLines);
    49.                 return 0;
    50.             case "-v":
    51.             case "--version":
    52.                 String version = EnvironmentInformation.getVersion();
    53.                 String commitID = EnvironmentInformation.getRevisionInformation().commitId;
    54.                 System.out.print("Version: " + version);
    55.                 System.out.println(
    56.                         commitID.equals(EnvironmentInformation.UNKNOWN)
    57.                                 ? ""
    58.                                 : ", Commit ID: " + commitID);
    59.                 return 0;
    60.             default:
    61.                 System.out.printf("\"%s\" is not a valid action.\n", action);
    62.                 System.out.println();
    63.                 System.out.println(
    64.                         "Valid actions are \"run\", \"run-application\", \"list\", \"info\", \"savepoint\", \"stop\", or \"cancel\".");
    65.                 System.out.println();
    66.                 System.out.println(
    67.                         "Specify the version option (-v or --version) to print Flink version.");
    68.                 System.out.println();
    69.                 System.out.println(
    70.                         "Specify the help option (-h or --help) to get help on the command.");
    71.                 return 1;
    72.         }
    73.     } catch (CliArgsException ce) {
    74.         return handleArgException(ce);
    75.     } catch (ProgramParametrizationException ppe) {
    76.         return handleParametrizationException(ppe);
    77.     } catch (ProgramMissingJobException pmje) {
    78.         return handleMissingJobException();
    79.     } catch (Exception e) {
    80.         return handleError(e);
    81.     }
    82. }

    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.");
    8.     //获取run的动作，默认的配置项
    9.     final Options commandOptions = CliFrontendParser.getRunCommandOptions();
    10.     //根据用户指定的配置项，进行解析 例如-t -p -c等
    11.     final CommandLine commandLine = getCommandLine(commandOptions, args, true);
    12.  
    13.     // evaluate help flag,只要参数中包含help 打印后结束
    14.     if (commandLine.hasOption(HELP_OPTION.getOpt())) {
    15.         CliFrontendParser.printHelpForRun(customCommandLines);
    16.         return;
    17.     }
    18.  
    19.     final CustomCommandLine activeCommandLine =
    20.             validateAndGetActiveCommandLine(checkNotNull(commandLine));
    21.  
    22.     final ProgramOptions programOptions = ProgramOptions.create(commandLine);
    23.  
    24.     //获取 用户的jar包和其他依赖
    25.     final List jobJars = getJobJarAndDependencies(programOptions);
    26.  
    27.     //获取有效配置：HA的id、Target（session、per-job）、JobManager内存、TaskManager内存、每个TM的slot数
    28.     final Configuration effectiveConfiguration =
    29.             getEffectiveConfiguration(activeCommandLine, commandLine, programOptions, jobJars);
    30.  
    31.     LOG.debug("Effective executor configuration: {}", effectiveConfiguration);
    32.     // PackagedProgram 类很关键，
    33.     try (PackagedProgram program = getPackagedProgram(programOptions, effectiveConfiguration)) {
    34.         // 执行程序
    35.         executeProgram(effectiveConfiguration, program);
    36.     }
    37. }

    yarn per-job模式源码分析

    以样例TopSpeedWindowing.jar为例，命令行提交命令为：

    ./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar

    application模式源码分析

    Flink 1.11 版本引入了 Application 模式。

    应用模式是唯一一个main方法不在客户端上执行的。其他两种模式都要在本地管理依赖资源，运行main方法然后生成JobGraph并提交到集群，不仅增加了网络传输的压力，还消耗了客户端大量的CPU资源。为了解决这个问题，应用模式先把用户jar包等资源提交到资源平台，然后创建Flink集群 并 自动在服务器上运行应用main方法。在main方法中可以通过execute或executeAysnc提交任务并记录提交作业id，作业执行完毕后，集群自动关闭。因此这种模式可以跟Per Job一样做到单任务的资源隔离，同时也可以解决客户端的计算瓶颈问题。

    另外，应用模式支持提交多个作业，作业的顺序依赖于启动顺序。如果使用的是execute()会阻塞按顺序执行；如果使用的是executeAysnc，各个任务可能会乱序执行。

    bin目录下 flink脚本内容，可以看到 入口类为 org.apache.flink.client.cli.CliFrontend

    1. #!/usr/bin/env bash
    2. ################################################################################
    3. #  Licensed to the Apache Software Foundation (ASF) under one
    4. #  or more contributor license agreements.  See the NOTICE file
    5. #  distributed with this work for additional information
    6. #  regarding copyright ownership.  The ASF licenses this file
    7. #  to you under the Apache License, Version 2.0 (the
    8. #  "License"); you may not use this file except in compliance
    9. #  with the License.  You may obtain a copy of the License at
    10. #
    11. #      http://www.apache.org/licenses/LICENSE-2.0
    12. #
    13. #  Unless required by applicable law or agreed to in writing, software
    14. #  distributed under the License is distributed on an "AS IS" BASIS,
    15. #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    16. #  See the License for the specific language governing permissions and
    17. # limitations under the License.
    18. ################################################################################
    19.  
    20. target="$0"
    21. # For the case, the executable has been directly symlinked, figure out
    22. # the correct bin path by following its symlink up to an upper bound.
    23. # Note: we can't use the readlink utility here if we want to be POSIX
    24. # compatible.
    25. iteration=0
    26. while [ -L "$target" ]; do
    27.     if [ "$iteration" -gt 100 ]; then
    28.         echo "Cannot resolve path: You have a cyclic symlink in $target."
    29.         break
    30.     fi
    31.     ls=`ls -ld -- "$target"`
    32.     target=`expr "$ls" : '.* -> \(.*\)$'`
    33.     iteration=$((iteration + 1))
    34. done
    35.  
    36. # Convert relative path to absolute path
    37. bin=`dirname "$target"`
    38.  
    39. # get flink config
    40. . "$bin"/config.sh
    41.  
    42. if [ "$FLINK_IDENT_STRING" = "" ]; then
    43.         FLINK_IDENT_STRING="$USER"
    44. fi
    45.  
    46. CC_CLASSPATH=`constructFlinkClassPath`
    47.  
    48. log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-client-$HOSTNAME.log
    49. log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlog4j.configurationFile=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
    50.  
    51. # Add Client-specific JVM options
    52. FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_CLI}"
    53.  
    54. # Add HADOOP_CLASSPATH to allow the usage of Hadoop file systems
    55. exec "${JAVA_RUN}" $JVM_ARGS $FLINK_ENV_JAVA_OPTS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"

    使用application方式提交样例作业

    hadoop1 jps

    hadoop2 jps

    hadoop3 jps

    jobmanager

    taskmanager -p 3 1slot/taskmanager 所以 需要启动3个container

  • 相关阅读:
    【springboot】你了解@Autowired 和 @Resource吗？@Autowired 和 @Resource深入分析
    Android 摇一摇功能实现，重力加速度大于15
    [动态规划] (十四) 简单多状态 LeetCode LCR 091.粉刷房子
    【毕业设计】机器学习的员工离职模型研究-python
    Vue2详解
    【leetcode】排序数组中两个数字之和
    《第一行代码Andorid》阅读笔记-第一章
    爬虫基础入门
    基本概念 I 和 Q：I/Q 数据的基础知识
    无良软件测试培训机构，退退退
  • 原文地址：https://blog.csdn.net/yelangshisan/article/details/133265734