Flink任务提交源码阅读（一）：flink脚本

缺乏、安全感 2023-06-14 12:59 11阅读 0赞

### 文章目录 ###

*   *  Flink版本&任务提交方式及示例
     *  阅读继续：

PS：在工作中有个需求：使用java API的方式提交flink job 到yarn上，有yarn负责部署集群运行任务，待任务运行完成后销毁集群，官方没有提供类似的java API怎么办，没办法只能看源码自己干了，也趁这个机会阅读flink提交任务的源码并记录以便后期学习查看。

## Flink版本&任务提交方式及示例 ##

阅读版本：Flink1.9.0  
任务提交方式：运行命令行flink脚本  
使用flink脚本提交任务示例：

./bin/flink run -m yarn-cluster -yjm 2048 -ytm 4096 -ys 16  -p 20 -ynm flinkOnYarnTESTweb ./examples/batch/WordCount.jar

run 是flink 运行的一个action，后面是一堆参数。  
先来看看安装好flink1.9.0后，它的bin目录下有什么  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70]  
这么多脚本，我们暂时只关注flink、config.sh两个脚本文件；config.sh中定义了很多flink程序中会用到的环境变量，flink是任务提交的入口，在flink脚本中会引用config.sh  
看看flink脚本中都有啥：

# 获取当前脚本名
    target="$0"
    iteration=0
    # 判断脚本名合法性
    while [ -L "$target" ]; do
        if [ "$iteration" -gt 100 ]; then
            echo "Cannot resolve path: You have a cyclic symlink in $target."
            break
        fi
        ls=`ls -ld -- "$target"`
        target=`expr "$ls" : '.* -> $.*$$'`
        iteration=$((iteration + 1))
    done
    # 获取当前脚本的上一层路径
    bin=`dirname "$target"`
    # get flink config
    . "$bin"/config.sh
    if [ "$FLINK_IDENT_STRING" = "" ]; then
            FLINK_IDENT_STRING="$USER"
    fi
     
    CC_CLASSPATH=`constructFlinkClassPath`
     
    log=$FLINK_LOG_DIR/flink-$FLINK_IDENT_STRING-client-$HOSTNAME.log
    log_setting=(-Dlog.file="$log" -Dlog4j.configuration=file:"$FLINK_CONF_DIR"/log4j-cli.properties -Dlogback.configurationFile=file:"$FLINK_CONF_DIR"/logback.xml)
     
    # Add HADOOP_CLASSPATH to allow the usage of Hadoop file systems
    # 使用exec启用新进程，调用java方法org.apache.flink.client.cli.CliFrontend，这个类的main方法就是job提交的入口
    exec $JAVA_RUN $JVM_ARGS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.client.cli.CliFrontend "$@"

以上脚本可以解读为：在运行job提交的入口类CliFrontend之前为其设置必要的环境变量、参数及必要的jar包，其中尤为重要的一个jar包是：flink-dist\_2.11-1.9.0.jar  
所以执行flink脚本后最后运行的命令是这样的

exec /opt/jdk1.8.0_211/bin/java 
    -Dlog.file=/opt/flink-1.9.0/log/flink-root-client-louisvv.log 
    -Dlog4j.configuration=file:/opt/flink-1.9.0/conf/log4j-cli.properties 
    -Dlogback.configurationFile=file:/opt/flink-1.9.0/conf/logback.xml 
    -classpath /opt/flink-1.9.0/lib/log4j-1.2.17.jar
    :/opt/flink-1.9.0/lib/slf4j-log4j12-1.7.15.jar
    :/opt/flink-1.9.0/lib/flink-dist_2.11-1.9.0.jar
    ::: org.apache.flink.client.cli.CliFrontend

其中我们需要重点关注的是：  
1、加载flink classpath  
加载的三个jar包是flink程序运行必须设定的包，  
PS：后面我在工作中想要通过java API提交job的实现过程中，因为flink-dist\_2.11-1.9.0.jar包的问题卡了好久好久。  
2、程序的入口类：org.apache.flink.client.cli.CliFrontend类  
这个类是运行命令行命令的入口类，主要功能是接收并解析命令行传入的命令，调用相应工具类执行命令，它有一下actions：

*  run：编译并运行程序
 *  cancel：取消正在运行的程序（官方不推荐使用该方式）
 *  stop：使用保存点停止正在运行的程序（仅用于流作业）
 *  savepoint：触发正在运行的作业的保存点或处置现有的保存点
 *  info：显示程序执行计划（JSON）
 *  list：列出正在运行和计划的程序

我们使用**run**来提交job程序。

## 阅读继续： ##

[Flink1.9.0任务提交源码阅读（二）：入口类CliFrontend][Flink1.9.0_CliFrontend]  
[Flink1.9.0任务提交源码阅读（三）：Job提交 run()][Flink1.9.0_Job_ run]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191119203041450.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzE2MTgxMQ==,size_16,color_FFFFFF,t_70
[Flink1.9.0_CliFrontend]: https://blog.csdn.net/weixin_43161811/article/details/103151644
[Flink1.9.0_Job_ run]: https://blog.csdn.net/weixin_43161811/article/details/103152867