[1101]flink常用参数说明

文章目录

flink命令
- 查看flink任务
- flink run参数
- flink run -m yarn-cluster参数
- flink cancel
- flink stop：仅仅针对Streaming job
- flink modify 修改任务并行度
- flink savepoint
- 例子
- 获取最近savepoints
- flink任务重启步骤样例
yarn命令
- 查看yarn中的作业
- 杀掉yarn session
- 查看yarn的环境变量设置
master节点启动一个 Socket 服务

flink命令

查看flink任务

flink list：列出flink提交任务的job列表。
flink list -r/--runing：列出正在运行的job
flink list -s/--scheduled：列出已调度完成的job# 查看最近取消的是哪个命令
./bin/flink list -a

flink run参数

flink run命令执行模板：flink run [option]

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

flink run -m yarn-cluster参数

参数	参数全称	描述
-m	–jobmanager	yarn-cluster集群
-yd	–yarndetached	后台
-yjm	–yarnjobManager	jobmanager的内存
-ytm	–yarntaskManager	taskmanager的内存
-yn	–yarncontainer	TaskManager的个数
-yid	–yarnapplicationId	job依附的applicationId
-ynm	–yarnname	application的名称
-ys	–yarnslots	分配的slots个数

例：flink run -m yarn-cluster -yd -yjm 1024m -ytm 1024m -ynm -ys 1

flink cancel

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

也可以通过 UI 界面 cancel

flink stop：仅仅针对Streaming job

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

通过 -m 来指定要停止的 JobManager 的主机地址和端口

例： bin/flink stop -m 127.0.0.1:8081 d67420e52bd051fae2fddbaa79e046bb

取消和停止（流作业）的区别如下：

cancel() 调用，立即调用作业算子的 cancel() 方法，以尽快取消它们。如果算子在接到 cancel() 调用后没有停止，Flink 将开始定期中断算子线程的执行，直到所有算子停止为止。

stop() 调用，是更优雅的停止正在运行流作业的方式。stop() 仅适用于 Source 实现了 StoppableFunction 接口的作业。当用户请求停止作业时，作业的所有 Source 都将接收 stop() 方法调用。直到所有 Source 正常关闭时，作业才会正常结束。这种方式，使作业正常处理完所有作业。

flink modify 修改任务并行度

flink modify <job_id> [options]

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

例： flink modify -p 并行数 <job_pid>

flink savepoint

flink savepoint [options] <job_id>

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]

savepoint 与 checkpoint 的区别

checkpoint是增量做的，每次的时间短，数据量小，只要在程序里面启用后会自动触发，用户无需感知；savepoint是全量做的，时间长，数据量大，需要用户主动触发。

checkpoint 是作业failover 的时候自动使用，不需要用户指定，savepoint 一般用于程序版本更新、bug修复、A/B Test 等场景，需要用户指定。

例子

实时风控:对应的数据源mq有八个队列，所以并行度设置为8

flink run -m yarn-cluster -p 8 -ytm 1024m -yjm 2048m -ys 1 -c com.enmonster.OrderDataProcessing sdb-ms-1.0-SNAPSHOT.jar /opt/cloudera/parcels/FLINK-1.9.0-csa1.0.0.0-cdh6.3.0/lib/flink/plugins/config.txt

编号	参数	描述
1	flink run -m yarn-cluster	flink任务在Flink job on YARN模式下的固定参数
2	-p	指定的程序并行度
3	-ytm	每个TaskManger分配的内存大小
4	-yjm	JobManger分配的内存大小，一个程序一般只有一个JobManger
5	-ys	一个TaskManger中启动的slot个数
6	-s	程序checkpoint检查点手动恢复指定路径，只在程序重新启动的时候使用
7	-c	程序主类类名
8	sdb-ms-1.0-SNAPSHOT.jar	程序jar包所在位置
9	/opt/cloudera/parcels/FLINK-1.9.0-csa1.0.0.0-cdh6.3.0/lib/flink/plugins/	程序配置文件的存放路径
10	config.txt	程序的配置文件

获取最近savepoints

hdfs dfs -ls hdfs:///user/flink/savepoints

flink任务重启步骤样例

#取消任务，得到savepoints地址
flink cancel a2c986adbc9ed9ec646fa0d926439775 -s hdfs:/user/flink/savepoints -yid application_1596180472513_1706#savepoints地址：
hdfs://cdh138.enmonster.org:8020/user/flink/savepoints/savepoint-a2c986-50aaa089b02f#kill任务
yarn application -kill application_1596180472513_1706#重启任务
flink run -m yarn-cluster -p 8 -ytm 1024m -yjm 2048m -ys 1 -ynm SDB_JOB -yD log4j2.formatMsgNoLookups=true -s hdfs://cdh138.enmonster.org:8020/user/flink/savepoints/savepoint-a2c986-50aaa089b02f -c com.enmonster.OrderDataProcessing sdb-ms-1.0-SNAPSHOT.jar /opt/cloudera/parcels/FLINK-1.9.0-csa1.0.0.0-cdh6.3.0/lib/flink/plugins/config.txt

log4j2修复bug参数

-yD log4j2.formatMsgNoLookups=true

yarn命令

查看yarn中的作业

yarn application --listyarn application -list -appStates ALL  #查看所有的作业#看某个作业的详细信息:
yarn application -status application_1565872581822_5969#查看相应状态的作业:
yarn application --list -appStates FINISHED >1.txtcat 1.txt |grep 2019-08-11状态有:
state can be one of the following:
ALL,NEW,NEW_SAVING,SUBMITTED,ACCEPTED,RUNNING,FINISHED,FAILED,KILLED

杀掉yarn session

yarn appclication -kill application_1625727533919_9502#kill掉default队列中的任务
for i in  `yarn application  -list | grep -w  default | awk '{print $1}' | grep application_`; do yarn  application -kill $i; done

查看yarn的环境变量设置

#查看yarn的环境变量设置:
yarn envvars
WARNING: YARN_OPTS has been replaced by HADOOP_OPTS. Using value of YARN_OPTS.
JAVA_HOME='/usr/java/jdk1.8.0_181-cloudera'
HADOOP_YARN_HOME='/opt/cloudera/parcels/CDH-6.2.0-1.cdh6.2.0.p0.967373/lib/hadoop/libexec/../../hadoop-yarn'
YARN_DIR='./'
YARN_LIB_JARS_DIR='lib'
HADOOP_CONF_DIR='/etc/hadoop/conf'
HADOOP_TOOLS_HOME='/opt/cloudera/parcels/CDH-6.2.0-1.cdh6.2.0.p0.967373/lib/hadoop'
HADOOP_TOOLS_DIR='share/hadoop/tools'
HADOOP_TOOLS_LIB_JARS_DIR='share/hadoop/tools/lib'

master节点启动一个 Socket 服务

命令：nc -lk 8888（提示：nc: command not found，通过yum install nc来安装）

flink.apache：https://flink.apache.org/zh/flink-architecture.html

参考：https://blog.csdn.net/l_521mu/article/details/114327993
https://blog.csdn.net/hxbao/article/details/107982264
https://blog.csdn.net/Black_8/article/details/107863652
https://www.cnblogs.com/hongfeng2019/p/12323655.html

flink学习项目：
https://github.com/QuakeWang/FlinkTutorial
https://github.com/mys-mx/Flink-study/tree/master/Flink_study/data

Flink常见异常和错误信息小结：https://blog.csdn.net/u013411339/article/details/105401066/