flink1.12.x flink run 执行flink程序(per-job cluster模式)

更详细解释:
https://blog.csdn.net/Zsigner/article/details/107787344

查看所有参数

执行下面命令,可以查看所有参数:

flink run -help

常用参数解释

flink run

参数:

-c,–class Flink应用程序的入口。jar包中没有配置 Main-Class,启动时需指定主启动类

-p,–parallelism 执行并行度

per-job cluster模式 的参数

常用参数:

-m yarn-cluster 指定提交到哪台主机上,yarn-cluster 表示提交到 yarn 集群上
-yjm,–yarnjobManagerMemory jobmanager的内存(单位M)
-ytm,–yarntaskManagerMemory 每个taskmanager的内存

-yqu,--yarnqueue <arg>  指定YARN queue
  • -m:指定提交到哪台主机上
  • yarn-cluster:表示提交到 yarn 集群上

其他参数:

-d,–detached 是否使用分离模式
-yat,–yarnapplicationType 设置yarn应用的类型
-yD <property=value> 使用给定属性的值
-yd,–yarndetached 使用yarn分离模式
-yh,–yarnhelp yarn session的帮助
-yid,–yarnapplicationId 挂到正在运行的yarnsession上
-yj,–yarnjar Flink jar文件的路径
-ynl,–yarnnodeLabel 指定 YARN 应用程序 YARN 节点标签
-ynm,–yarnname 自定义yarn应用名称
-yq,–yarnquery 显示yarn的可用资源
-ys,–yarnslots 指定每个taskmanager的slots数
-yt,–yarnship 在指定目录中传输文件
-yz,–yarnzookeeperNamespace 用来创建ha的zk子路径的命名空间
-z,–zookeeperNamespace 用来创建ha的zk子路径的命名空间

例子

执行自带的 WordCount jar包:

./bin/flink run -m yarn-cluster -p 2 ./examples/batch/WordCount.jar      \
--input hdfs://hadoop1:8020/test2/data.txt                        \
--output hdfs://hadoop1:8020/flink_result/1

注意:默认需要 6 个slots(如下图),但只配置了 2-4 个slots,导致申请不下来足够的slot,可能 无法运行

例子:仅指定 slots

./bin/flink run -m yarn-cluster -p 2   -ys 1   ./examples/batch/WordCount.jar   \
--input hdfs://hadoop1:8020/test2/data.txt  \           --output hdfs://hadoop1:8020/flink_result/5

例子:指定内存大小 和 slots

./bin/flink run -m yarn-cluster -p 2  -yjm 1000 -ytm 1000  -ys 1   ./examples/batch/WordCount.jar  \
--input hdfs://hadoop1:8020/test2/data.txt  \           --output hdfs://hadoop1:8020/flink_result/4

注意:内存不能小于 1000,否则可能内存不够,无法运行

例子:指定yarn队列

flink run -yqu root.default  -m yarn-cluster -yjm 1000 -ytm 1000  /program/flink-1.12.5/examples/batch/WordCount.jar   --input hdfs://hadoop1:8020/test2/data.txt --output hdfs://hadoop1:8020/flink_result12291419

例子:执行自己写的程序

./bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 -c top.malaoshi.TestDataStream2_lambda2_hdfs /program/test_flink.jar --output hdfs://hadoop1:8020/test_result/1

设置批处理模式

默认是 流处理模式,通过参数可以设置批处理模式,如下:

flink run -Dexecution.runtime-mode = BATCH

原文出处:/show_1IX4D76vjB35.html