Flink源码--CLI提交Job

清疚 2023-05-30 09:03 24阅读 0赞

本文链接： [https://blog.csdn.net/lmalds/article/details/70174778][https_blog.csdn.net_lmalds_article_details_70174778]

Flink提供了一个命令行接口，来运行并控制打包后的jar文件中的程序。其是Flink安装的一部分，在本地以及分布式模式下都可以使用。CLI命令接口位于:

<flink_home>/bin/flink
     
      
     
      
       1

当提交Job后，其默认会连接到Flink的JobManager。

提交Flink的Job有一个前提，即JobManager必须处于运行中，Flink有3种运行运行模式：

1、本地模式：<flink-home>/bin/start-local.sh
    
    2、集群模式：<flink-home>/bin/start-cluster.sh
    
    3、Yarn或Mesos环境
     
      
     
      
       1
       2
       3
       4
       5

## 1、使用语法 ##

具体的语法参见：[Command-Line Interface][]。

这里可将CLI的命令抽象为：

./flink <ACTION> [OPTIONS] [ARGUMENTS]
     
      
     
      
       1

## 2、flink脚本 ##

脚本文件：$FLINK\_HOME/bin/flink

其主要是执行. “$bin”/config.sh来加载flink的环境配置信息，而config.sh则读取flink-conf.yaml、slaves、masters等来读取配置。

在flink脚本文件的最后（55行），执行了最终的Job：

exec $JAVA_RUN $JVM_ARGS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.CliFrontend "$@"
     
      
     
      
       1

这里就是用java命令来执行flink的类：`org.apache.flink.client.CliFrontend` 来完成。

例如，我的CLI命令是：

flink run -c com.toptrade.Job toptrade-flink-1.0.jar prod.properties
     
      
     
      
       1

则flink做的事情是校验我的命令，并加载flink的环境配置、taskManagers、HA的机器配置、日志文件、ClassPath以及hadoop的配置等，并最终执行java运行命令，如下：

![这里写图片描述][SouthEast]

即指定：

1、/home/flink/java/jdk1.8.0_60/bin/java 
    2、-Dlog.file、-Dlog4j.configuration、-Dlogback.configurationFile
    3、-classpath
    4、执行org.apache.flink.client.CliFrontend类
    5、引用参数run -c com.toptrade.Job toptrade-flink-1.0.jar prod.properties
    6、加载flink环境配置信息
     
      
     
      
       1
       2
       3
       4
       5
       6

## 3、org.apache.flink.client.CliFrontend ##

此类是所有Job的入口，通过读取flink的环境、配置信息，并根据用户提供的jar包和入口类，进行最终的Job提交。

![这里写图片描述][SouthEast 1]

我们再来看看parseParameters(args)方法：  
![这里写图片描述][SouthEast 2]

这里会根据《Action》的不同，进行不同的操作，我们主要看下run，即程序的执行：

这里的run方法主要分为以下几大部分：

1、将命令行中的options、程序入口类、jar文件等封装起来

//命令行选项类，继承自ProgramOptions
            RunOptions options;
            try {
                // 解析args，将配置信息用RunOptions类封装起来
                options = CliFrontendParser.parseRunCommand(args);
            }
     
      
     
      
       1
       2
       3
       4
       5
       6

关于RunOptions，其继承关系图如下：

![这里写图片描述][SouthEast 3]

我们看看最基础的抽象类CommandLineOptions：  
![这里写图片描述][SouthEast 4]

下层的抽象类ProgramOptions，也是一个基础类，其将配置中的jar、入口类、classPath以及《options》的一些信息封装起来：  
![这里写图片描述][SouthEast 5]

RunOptions类：只是调用父类的构造方法：  
![这里写图片描述][SouthEast 6]

2、创建一个封装了入口类、jar文件、classpath路径、用户配置参数的实例：PackagedProgram

// 构建程序
            PackagedProgram program;
            try {
                LOG.info("Building program from JAR file");
                // 创建一个封装了入口类、jar文件、classpath路径、用户配置参数的实例：PackagedProgram
                program = buildProgram(options);
            }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7

buildProgram方法主要是加载并初始化PackagedProgram类：

// Get assembler class
            String entryPointClass = options.getEntryPointClassName();
    
            PackagedProgram program = entryPointClass == null ?
                    new PackagedProgram(jarFile, classpaths, programArgs) :
                    new PackagedProgram(jarFile, classpaths, entryPointClass, programArgs);
     
      
     
      
       1
       2
       3
       4
       5
       6

3、封装必要的函数并提交到远程集群中

// 3、封装必要的函数并提交到远程集群中，有2个子类：StandaloneClusterClient以及YarnClusterClient,现在也包含了Mesos的支持
            ClusterClient client = null;
            try {
                // 根据RunOptions和PackagedProgram，创建Client
                client = createClient(options, program);
                client.setPrintStatusDuringExecution(options.getStdoutLogging());
                client.setDetached(options.getDetachedMode());
                LOG.debug("Client slots is set to {}", client.getMaxSlots());
    
                LOG.debug(options.getSavepointRestoreSettings().toString());
    
                int userParallelism = options.getParallelism();
                LOG.debug("User parallelism is set to {}", userParallelism);
                if (client.getMaxSlots() != -1 && userParallelism == -1) {
                    logAndSysout("Using the parallelism provided by the remote cluster ("
                        + client.getMaxSlots()+"). "
                        + "To use another parallelism, set it at the ./bin/flink client.");
                    userParallelism = client.getMaxSlots();
                }
                // 最终的执行，根據PackagedProgram以及ClusterClient和并行度，执行程序
                return executeProgram(program, client, userParallelism);
            }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22

这里有2个主要的方法：

1、client = createClient(options, program);
    2、executeProgram(program, client, userParallelism);
     
      
     
      
       1
       2

我们分别来看下2个方法。  
createClient：创建ClusterClient对象，包含集群模式信息，JobManager地址以及WebUI端口等。  
![这里写图片描述][SouthEast 7]

最后也是最重要的一个方法：executeProgram。

// Client通过Actor提交程序到JobManager
        protected int executeProgram(PackagedProgram program, ClusterClient client, int parallelism) {
            //输出：Starting execution of program
            logAndSysout("Starting execution of program");
            // 包含JobId的Job提交类
            JobSubmissionResult result;
            try {
                //执行CluterClient中的run方法
                result = client.run(program, parallelism);
            }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10

继续看一下client.run(program, parallelism)：

public JobSubmissionResult run(PackagedProgram prog, int parallelism)
                throws ProgramInvocationException, ProgramMissingJobException
        {
            Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader());
            // 如果包含入口类（非交互模式提交Job）
            if (prog.isUsingProgramEntryPoint()) {
                // JobWithJars是一个Flink数据流计划，包含了jar中所有的类，以及用于加载用户代码的ClassLoader
                final JobWithJars jobWithJars;
                if (hasUserJarsInClassPath(prog.getAllLibraries())) {
                    jobWithJars = prog.getPlanWithoutJars();
                } else {
                    jobWithJars = prog.getPlanWithJars();
                }
                //跳转到3个参数的run方法
                return run(jobWithJars, parallelism, prog.getSavepointSettings());
            }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16

我们这里暂时只关注包含入口类的情况，不对交互提交模式进行分析。

// 此时Client已经连接到Flink的集群，该调用将被阻塞直到执行完成
        public JobSubmissionResult run(JobWithJars jobWithJars, int parallelism, SavepointRestoreSettings savepointSettings)
                throws CompilerException, ProgramInvocationException {
            ClassLoader classLoader = jobWithJars.getUserCodeClassLoader();
            if (classLoader == null) {
                throw new IllegalArgumentException("The given JobWithJars does not provide a usercode class loader.");
            }
            //这里根据流或批，为每一个operator进行优化，例如shuffle的方式，hash join、sort-merge join或者广播等进行优化
            OptimizedPlan optPlan = getOptimizedPlan(compiler, jobWithJars, parallelism);
            //根据优化后的执行计划，jar文件，classpath，类加载器，保存点设置运行
            return run(optPlan, jobWithJars.getJarFiles(), jobWithJars.getClasspaths(), classLoader, savepointSettings);
        }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12

public JobSubmissionResult run(FlinkPlan compiledPlan,
                List<URL> libraries, List<URL> classpaths, ClassLoader classLoader, SavepointRestoreSettings savepointSettings)
            throws ProgramInvocationException
        {
            // 生成JobGraph，并将JobGraph提交到JobManager
            JobGraph job = getJobGraph(compiledPlan, libraries, classpaths, savepointSettings);
            return submitJob(job, classLoader);
        }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8

关于如何生成JobGraph，这里忽略。

最后，我们看下submitJob方法，这里以StandaloneClusterClient.submitJob方法为例，判断是否是分离模式，如果不是的话，则运行ClusterClient的run(JobGraph jobGraph, ClassLoader classLoader)方法：

public JobExecutionResult run(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
    
            waitForClusterToBeReady();
    
            final LeaderRetrievalService leaderRetrievalService;
            try {
                // 根据配置信息中是否是高可用模式，创建LeaderRetrievalService对象
                leaderRetrievalService = LeaderRetrievalUtils.createLeaderRetrievalService(flinkConfig, true);
            } catch (Exception e) {
                throw new ProgramInvocationException("Could not create the leader retrieval service", e);
            }
    
            try {
                logAndSysout("Submitting job with JobID: " + jobGraph.getJobID() + ". Waiting for job completion.");
                // 调用JobClient的submitJobAndWait方法，提交Job
                this.lastJobExecutionResult = JobClient.submitJobAndWait(actorSystemLoader.get(), flinkConfig,
                    leaderRetrievalService, jobGraph, timeout, printStatusDuringExecution, classLoader);
                return this.lastJobExecutionResult;
            } catch (JobExecutionException e) {
                throw new ProgramInvocationException("The program execution failed: " + e.getMessage(), e);
            }
        }
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22

后续的流程主要是通过JobClient将Job提交给JobClient actor，而后由这个actor提交给JobManager，完成Job的提交。

## 4、日志输出 ##

![这里写图片描述][SouthEast 8]

可以看到，主要是在createClient(options, program)和executeProgram(program, client, userParallelism)方法中输出。

以上便是作业提交的全部流程，中间有些过程可能偏离了主线，而有些过程又一笔带过，很多细节没有涉及到。

关于如何生成StreamGraph以及JobGraph，可以参考群主的博客：

> [http://blog.csdn.net/yanghua\_kobe/article/details/51935158][http_blog.csdn.net_yanghua_kobe_article_details_51935158]  
> [http://blog.csdn.net/yanghua\_kobe/article/details/52006638][http_blog.csdn.net_yanghua_kobe_article_details_52006638]  
> [http://blog.csdn.net/yanghua\_kobe/article/details/68935547][http_blog.csdn.net_yanghua_kobe_article_details_68935547]  
> [http://blog.csdn.net/yanghua\_kobe/article/details/68953752][http_blog.csdn.net_yanghua_kobe_article_details_68953752]

[Link 1]: http://creativecommons.org/licenses/by-sa/4.0/
[https_blog.csdn.net_lmalds_article_details_70174778]: https://blog.csdn.net/lmalds/article/details/70174778
[Command-Line Interface]: https://ci.apache.org/projects/flink/flink-docs-release-1.2/setup/cli.html
[SouthEast]: https://img-blog.csdn.net/20170414153134794?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 1]: https://img-blog.csdn.net/20170414154234790?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 2]: https://img-blog.csdn.net/20170414154617052?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 3]: https://img-blog.csdn.net/20170414155445431?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 4]: https://img-blog.csdn.net/20170414155621560?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 5]: https://img-blog.csdn.net/20170414155808999?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 6]: https://img-blog.csdn.net/20170414160017808?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 7]: https://img-blog.csdn.net/20170414164506844?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 8]: https://img-blog.csdn.net/20170414174252328?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbG1hbGRz/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[http_blog.csdn.net_yanghua_kobe_article_details_51935158]: http://blog.csdn.net/yanghua_kobe/article/details/51935158
[http_blog.csdn.net_yanghua_kobe_article_details_52006638]: http://blog.csdn.net/yanghua_kobe/article/details/52006638
[http_blog.csdn.net_yanghua_kobe_article_details_68935547]: http://blog.csdn.net/yanghua_kobe/article/details/68935547
[http_blog.csdn.net_yanghua_kobe_article_details_68953752]: http://blog.csdn.net/yanghua_kobe/article/details/68953752