一、Yarn session（一般测试环境）
yarn session会初始启动指定的tm数量。 job提交后再指定的session 内运行。
其它job运行，如果资源不够，就会一直等待直到占用的job运行完毕释放资源。jm所有job共享
1 在任意几点启动yarn-session:
下述属性启动4个任务管理器。 每个任务管理器占用1024M内存，且每任务管理器分配2个SLOT
./bin/ -n 4 -tm 1024 -s 2
如下图， yarn模式下会随机找到一机器做为job manager，在master提交后， 去到cdh03节点作为job manager运行
注：上述第一个红框为 yarn 集群中的application id. 运行日志可以根据此ID，去yarn WEB UI中查看。
2 提交JOB：
在提交session的节点：
./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_.jar --input hdfs:///flinktest/ --output hdfs:///flinktest/output
如果是其它节点：需要指定 job管理器 -m ，否则会跑到 master节点 8081口提交任务，最后超时报错
./flink run -m cdh02:40990 -c WordCount /home/hadoop/flinktest/FlinkStream_.jar --input hdfs:///flinktest/ --output hdfs:///flinktest/output2 -m cdh02:409904099
也可以指定sessionid来提交特别是在多个session的情况下 -yid application_157118_00001
3 yarn 模式下日志查看跟 standalone方式有点不一样。在yarn WEB UI 可以查看
application id 可以根据步骤1 的ID，去yarn管理界面查看。
点击，ID连接，进去找到对应的am管理的容器，查看日志
物理日志路径， 根据yarn配置文件，在/yarn /container-logs 目录下（可以看到应该在cdh01节点）
3 停止job
yarn application -kill session id.二 yarn on per job(生产使用)
1 常规启动：
bin/flink run -c WordCount -m yarn-cluster examples/batch/， 注意参数 -m yarn-cluster
此命令直接向yarn申请资源，构建集群，每job 独占jobmanager运行。
tm数=max(子任务并行书)/每tm槽位数.
2 停止JOB使用检查点。 action: stop
当没有使用flie或db保存，即内存保存检查点时。要停机，担心数据丢失。可以使用下述方式，手动指定检查点保存路径。后续启动时，再恢复。
flink stop -p savepointdir <pid> -yid <yid> [-p savepoint路径]
./flink stop -p hdfs://nameservice1/tmp c0a428874f675891400651d47855e85a -yid application_1599550090524_0046
3 也可以直接使用savepoint命令。 action： savepoint
flink savepoint jobId [targetDirectory] [-yid yarnAppId]
此命令可以针对运行中的job执行 savepoint操作。
pid可以进入FLINK WEB UI 查看：
3 从检查点启动(同常规启动比较只是多了个参数-s)
./flink run -s hdfs://nameservice1/tmp/savepoint-c0a428-b246908b40f5 -c WordCount -m yarn-cluster -ynm tts_log_2hdfs examples/batch/
参考：Flink SavePoint的使用_lvwenyuan_1的博客-****博客_flink savepoint配置