[1131]Flink(1.13)命令行提交Job

Flink提供了yarn上运行的3模式&＃xff0c;分别为Application Mode, Session-Cluster和Per-Job-Cluster模式。
Yarn 模式会动态申请资源

请注意&＃xff0c;客户端需要YARN_CONF_DIR或HADOOP_CONF_DIR环境变量来读取YARN和HDFS配置。没配置的话&＃xff0c;就默认是 /etc/hadoop/conf。

在flink1.13以前&＃xff0c;lib下是有hadoop相关的jar包的&＃xff0c;但是1.13以后被独立出去了&＃xff0c;所以需要加上环境变量&＃xff0c;可以加在config.sh的开头。

export HADOOP_CLASSPATH&＃61;&＃96;hadoop classpath&＃96;
export YARN_CLASSPATH&＃61;&＃96;yarn classpath&＃96;


相关的属性配置都可以用-D来设置&＃xff0c;比如&＃xff1a;
应用程序名设置&＃xff1a;-Dyarn.application.name&＃61;xxx
内存设置&＃xff1a;-Dtaskmanager.memory.process.size&＃61;4096m

Per-Job-Cluster新老版本启动方法

独享集群&＃xff0c;提交之后由yarn现启集群。

老版本&＃xff08;<&＃61;1.10&＃xff09;

flink run -m yarn-cluster -c xxx xxx.jar


新版本&＃xff08;>&＃61;1.11&＃xff09;

flink run -t yarn-per-job -c xxx xxx.jar


提交一个flink job到yarn&＃xff08;Flink on Yarn Per-Job Mode&＃xff09;
该模式下也可以指定--detached参数&＃xff0c;指定了则一旦作业提交被yarn接受&＃xff0c;客户端将停止。

./bin/flink run -t yarn-per-job -p 4 \
-Djobmanager.memory.process.size&＃61;2048m \
-Dtaskmanager.memory.process.size&＃61;4096m \
./examples/batch/WordCount.jar


需要帮助则&＃xff1a;

./bin/flink run -help


有啥参数可以使用-D来指定&＃xff0c;不要使用Options for yarn-cluster mode:下的那些参数比如-ynm、-yjm、-ytm等&＃xff0c;不生效。

• 提交样例

flink run -t yarn-per-job \ //指定运行模式
-d -ynm FlinkRetention \ //指定在jobmanager里面显示的名字
-Dyarn.application.name&＃61;FlinkRetention \ // 指定在yarn上的application的名字
-c com.bigdata.etl.FlinkRetention // 入口类/data/bigdata/flink-dw/target/flink-dw.jar \ // 自己任务的jar包
--consumer.bootstrap.servers ${consumer_bootstrap_servers} \ 需要传入的参数
--producer.bootstrap.servers ${producer_bootstrap_servers} \
--retentionGroupId ${retentionGroupId} \
--flinkSeconds ${flinkSeconds} \
--redis.ip ${redis_ip} \
--redis.port ${redis_port} \
--redis.password ${redis_password}


Session-Cluster

先启动一个集群&＃xff0c;再将任务提交到上面

Per-Job-Cluster 与 Session-Cluster 区别

• Per-Job-Cluster &＃xff1a;
  1.多少个job就有多少个集群

• Session-Cluster &＃xff1a;
  1.会首先在flink上启一个集群&＃xff0c;所有的job都交给一个集群中。
  2.各个应用之间共享资源。
  3.适用于小任务量使用&＃xff0c;大任务不太适用。

• 工作中推荐使用哪一种&＃xff1f;
  官方推挤&＃xff0c;大家推荐 使用Per-Job-Cluster的方式。
  Session-Cluster 是资源共享&＃xff0c;job太多&＃xff0c;可能会出现一些问题&＃xff0c;但是Session-Cluster也并非一无是处&＃xff0c;若job类型一致就可以放到一个集群中。

• 启动

bin/yarn-session.sh -d


-d 以后台的方式启动

• 启动日志

luster has been started in detached mode. In order to stop Flink gracefully, use the following command:
$ echo "stop" | ./bin/yarn-session.sh -id application_1628336315656_0001
If this should not be possible, then you can also kill Flink via YARN&＃39;s web interface or via:
$ yarn application -kill application_1628336315656_0001
Note that killing Flink might not clean up all job artifacts and temporary files.


• 日志说明

以 后台的方式启动&＃xff1a;luster has been started in detached mode.为了优雅地停止Flink&＃xff0c;请使用以下命令: $ echo "stop" | ./bin/yarn-session.sh -id application_1628336315656_0001如果这是不可能的&＃xff0c;那么你也可以通过YARN的web界面或通过以下方式杀死Flink:
$ yarn application -kill application_1628336315656_0001注意&＃xff0c;杀死Flink可能不会清除所有作业工件和临时文件。

• yarn

Application

应用模式&＃xff0c;1.11 之后次才有的&＃xff0c;类比 Per-Job-Cluster&＃xff0c;也是独享集群。

Per-Job-Cluster 与 Application 区别

就在于解析代码的位置&＃xff0c;

• Per-Job-Cluster&＃xff1a;
  解析main方法是在提交节点的本地。

• Application &＃xff1a;
  解析main方法是在Master。

高可用配置

• 更改yarn配置
  在yarn-site.xml中配置

yarn.resourcemanager.am.max-attempts4The maximum number of application master execution attempts.


• 更改flink配置
  在flink-conf.yaml中配置

yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-yarn


如果 yarn的application&＃xff0c;如果开启了高可用&＃xff0c;所有的jobid都会变成0000000000000000000&＃xff1b;官方的解释&＃xff0c;application模式不允许记录多个状态&＃xff0c;这样jobid就不好给了。

yarn的session集群&＃xff0c;如果开启了高可用&＃xff0c;-D应用id参数&＃61;appId&＃xff0c;指定提交到session集群&＃xff0c;可能会有问题。解决方式 &＃xff1a;不写 -t yarn-session

Yarn 高可用和Standalone 高可用区别

• Standalone 高可用
  同时启动多个JobManger&＃xff0c;若执行任务的JobManger挂了&＃xff0c;其他JobManager马上补上。

• Yarn 高可用
  JobManager只有一个&＃xff0c;但是配置了重试次数&＃xff0c;挂了之后再重启&＃xff08;重启一个新的ApplicationManager&＃xff09;&＃xff0c;利用了ApplicationMaster的重启机制。若在一定时间范围内重试都启动不了&＃xff0c;那么就真的挂了。如30秒内重试3次&＃xff0c;若重启成功&＃xff0c;则重试次数清零。

查看flink提交任务

# 查看flink提交任务
./bin/flink list# 查看最近取消的是哪个命令
./bin/flink list -a


参考&＃xff1a;https://blog.51cto.com/u_15318160/3248006
https://blog.csdn.net/chanyue123/article/details/110442617
https://www.jianshu.com/p/861bc86465c9
https://www.jianshu.com/p/10f0d4ff50be
https://blog.csdn.net/qq_27474277/article/details/116663318