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的记录。

如：

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界面显示结果：

---

二、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

输出结果：

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 

执行结果：

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>

---

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)]