Flink的多种提交方式

痛定思痛。 2022-01-26 16:04 401阅读 0赞

## Flink的多种提交/运行方式 ##

Flink具有多种提交方式，比如：常用的local模式，stantalone模式，yarn模式，k8s等。

这里主要对比local，stantalone，yarn三种提交方式。

### 一、本地（local）模式，仅开发使用 ###

#### 1.1 纯粹的local模式运行 ####

这就是直接运行项目中的代码的方式提交：

public class TestLocal {
        public static void main(String[] args) throws Exception {
    
            // getExecutionEnvironment()方法可以根据flink运用程序如何提交判断出是那种模式提交，Local本地提交，Cluster标是standalone提交，Yarn提交好像是YarnCluster
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            // ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
            env.setParallelism(1); // 设置并行度为1
    
            env
                    .readTextFile("D:\\code\\scalacode\\flinkstreamingreportforms\\pom.xml")
                    .print();
        }
    }

结果就是

#### 1.2 local使用remote的方式运行（不建议） ####

一般可以使用这种模式进行远程debug。如：

public class TestLocal {
        public static void main(String[] args) throws Exception {
    
            ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment("remote_ip", 8090, "D:\\code\\scalacode\\flinkstreamingreportforms\\target\\flink-streaming-report-forms-1.0-SNAPSHOT-jar-with-dependencies.jar");
    
            System.out.println(env.getParallelism());
    
            env
                    .readTextFile("hdfs://remote_ip:9000/tmp/test.txt")
                    .print();
        }
    }

上面提交给的我们的开发环境flink集群。在flink集群的web ui会存在Running Job/Compaleted Jobs的记录。

如：

![image][]

#### 1.3 本地提交到remote集群 ####

项目代码：

public class TestLocal {
        public static void main(String[] args) throws Exception {
            // getExecutionEnvironment()方法可以根据flink运用程序如何提交判断出是那种模式提交，Local本地提交，Cluster标是standalone提交，Yarn提交好像是YarnCluster
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            
            env
                    .readTextFile("hdfs://remote_ip:9000/tmp/test.txt")
                    .print();
        }
    }

将项目打成jar包，将使用flink run 指定集群的模式提交（需要在local机器上，有flink提交命令）。如：

# -m flink 集群地址
    # -p 配置job的并行度
    # -c Job执行的main class
    ./flink.bat run -m remote_ip:8090 -p 1 -c com.test.TestLocal ../flink-streaming-report-forms-1.0-SNAPSHOT-jar-with-dependencies.jar

执行结果如下：

\[外链图片转存失败(img-n92W3e4w-1563439836745)(https://raw.githubusercontent.com/zhang3550545/image\_center/master/image-2019/flink-run-02.png)\]

也会在flink web ui界面显示结果：

![image][image 1]

--------------------

### 二、standalone ###

上面讲了flink在local机器上进行提交，需要指定flink的master信息。

standalone模式提交也是类似，不过可以不用指定master节点，还有个区别就是，提交是在flink集群的机器节点上。

可能有人问，在local机器上提交与flink机器节点上提交有啥区别？

区别不太，主要是，提交后，会有一些log信息输出，这些信息在哪里记录或展示，还有就是程序最后会收集结果collect，最后的收集结果也就在对应提交的机器上。

下面介绍flink的前台运行与后台运行，以及常用参数：

以下均已1.3打出来的jar作为测试。

#### 2.1 flink run 前台运行 ####

/opt/flink/bin/flink run -p 1 -c com.test.TestLocal ./flink-streaming-report-forms-1.0-SNAPSHOT-jar-with-dependencies.jar

输出结果：

![image][image 2]

#### 2.2 flink run 后台运行 ####

因1.3代码是直接print出来，使用后台模式会报错，这边修改代码，直接写入hdfs，再使用后台模式提交。

# 通过 -d 表示后台执行
    /opt/flink/bin/flink run -p 1 -c com.test.TestLocal -d ./flink-streaming-report-forms-1.0-SNAPSHOT-jar-with-dependencies.jar

执行结果：

![image][image 3]

#### 2.3 flink run 命令常用的参数： ####

flink run命令参数如下：

flink run命令执行模板：`flink run [option] <jar-file> <arguments>`

-c,--class <classname> : 需要指定的main方法的类
    
    -C,--classpath <url> : 向每个用户代码添加url，他是通过UrlClassLoader加载。url需要指定文件的schema如（file://）
    
    -d,--detached : 在后台运行
    
    -p,--parallelism <parallelism> : job需要指定env的并行度，这个一般都需要设置。
    
    -q,--sysoutLogging : 禁止logging输出作为标准输出。
    
    -s,--fromSavepoint <savepointPath> : 基于savepoint保存下来的路径，进行恢复。
    
    -sas,--shutdownOnAttachedExit : 如果是前台的方式提交，当客户端中断，集群执行的job任务也会shutdown。

\[外链图片转存失败(img-YIQdSvzH-1563439836747)(https://raw.githubusercontent.com/zhang3550545/image\_center/master/image-2019/flink-run-06.png)\]

--------------------

### 三、flink on yarn ###

flink on yarn提交有2种方式，第一种是`yarn-session`，一种是`flink run -m yarn-cluster`。下面分别介绍2种方式的用法。

#### 3.1 yarn-session ####

yarn-session提交的方式有3步：

第一步：分配资源，生成application\_id

`yarn-session -jm 1024m -nm flinkOnYarnTest -s 1 -tm 1024m -d`

第二步：yarn-session依附application\_id

`yarn-session -id/--applicationId <application_id>`

第三步：flink 提交job

`flink run -c classname <jar> <arguments>`

下面是yarn-session的命令行参数

\[外链图片转存失败(img-0hyI6BGC-1563439836748)(https://raw.githubusercontent.com/zhang3550545/image\_center/master/image-2019/flink-run-07.png)\]

#### 3.2 flink run -m yarn-cluster ####

`flink run -m yarn-cluster` 实际上和上面提交standalone的方式一样。不过对于yarn的资源的命令参数有区别，如下：

# -m/--jobmanager : yarn-cluster集群
    # -yd/--yarndetached : 后台
    # -yjm/--yarnjobManager : jobmanager的内存
    # -ytm/--yarntaskManager : taskmanager的内存
    # -yid/--yarnapplicationId : job依附的applicationId
    # -ynm/--yarnname : application的名称
    # -ys/--yarnslots : 分配的slots个数
    
    flink run -m yarn-cluster -yd -yjm 1024m -ytm 1024m -ynm <name> -ys 1 <jar> <arguments>

![image][image 4]

--------------------

#### flink的其他命令 ####

1.flink list

flink list：列出flink的job列表。
    
    flink list -r/--runing :列出正在运行的job
    
    flink list -s/--scheduled :列出已调度完成的job

2.flink cancel

flink cancel [options] <job_id> : 取消正在运行的job id
    
    flink cancel -s/--withSavepoint <path> <job_id> ： 取消正在运行的job，并保存到相应的保存点

3.flink stop：仅仅针对Streaming job

flink stop [options] <job_id>
    
    flink stop <job_id>：停止对应的job

4.flink modify

flink modify <job_id> [options] 
    
    flink modify <job_id> -p/--parallelism p : 修改job的并行度

**5.flink savepoint**（重要）

flink savepoint [options] <job_id> <target directory>
    
    eg:
    
    # 触发保存点
    flink savepoint <job_id> <hdfs://xxxx/xx/x> : 将flink的快照保存到hdfs目录
    
    # 使用yarn触发保存点
    flink savepoint <job_id> <target_directory> -yid <application_id>
    
    # 使用savepoint取消作业
    flink cancel -s <tar_directory> <job_id>
    
    
    
    # 从保存点恢复
    flink run -s <target_directoey> [:runArgs]
    
    # 如果复原的程序，对逻辑做了修改，比如删除了算子可以指定allowNonRestoredState参数复原。
    flink run -s <target_directory> -n/--allowNonRestoredState [:runArgs]

\[外链图片转存失败(img-SE862DmC-1563439836749)(https://raw.githubusercontent.com/zhang3550545/image\_center/master/image-2019/flink-run-09.png)\]

[image]: /images/20220126/2859ba553d1448be904c5a45bc906de9.png
[image 1]: /images/20220126/015f1489aa9a4be2bd3f58a611fa2449.png
[image 2]: /images/20220126/a5fa058922d445c4b896b61a60834ab1.png
[image 3]: /images/20220126/2b2a5d94487d46a6a108ff59e76d3f3b.png
[image 4]: /images/20220126/25f2e38719ff4205b40a5245f5bc2fd5.png