flink on yarn模式

在flink on yarn模式中，flink yarn-session的两种提交方式

两种提交方式

1.yarn-session为flink app开辟公用资源

在yarn中初始化一个flink集群，开辟指定的资源，以后提交任务都向这里提交。这个flink集群会常驻在yarn集群中，除非手工停止。
2.每个job提供一个yarn-session
每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。

第一种方式：

1.首先启动yarn session，并且会启动Flink的两个必要服务：JobManager和TaskManagers，然后你可以向集群提交作业。同一个Session中可以提交多个Flink作业。

我们先看看这个脚本支持哪些参数：

./bin/yarn-session.sh
Usage:
   Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers) 申请多少资源
   Optional
     -D <arg>                    Dynamic properties动态属性
     -d,–detached              Start detached类似于spark的集群模式spark-cluster:后台运行，默认类似与spark-client
     -f6,--flip6                     Specify this option to start a Flip-6 Yarn session cluster.
     -id,--applicationId <arg>       Attach to running YARN session
     -jm,--jobManagerMemory <arg>    Memory for JobManager Container [in MB] 
     -nm,--name<arg>             Set a custom name for the application on YARN
     -q,–query                Display available YARN resources (memory, cores)
     -qu,--queue <arg>            Specify YARN queue.
     -s,--slots  <arg>           Number of slots per TaskManager 没个TaskManager 中的线程数量（内核数）
     -st,--streaming             Start Flink in streaming mode 
     -tm,--taskManagerMemory <arg>    Memory per TaskManager Container [in MB]
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for high availability mode

线上脚本： bin/yarn-session.sh -n 7 -s 8 -jm 3072 -tm 32768 -qu root..-nm - -d
其中申请7个taskManager 每个8核 每个taskmanager有32768M内存 ,后台运行
我的在yarn申请的flink资源命令：

HADOOP_HOME=/hadoop/hadoop-2.7.2/ HADOOP_CONF_DIR=/hadoop/hadoop-2.7.2/etc/hadoop/ ./yarn-session.sh -nm "Flink test" -n 15 -jm 8192 -tm 8192

此时我在yarn我申请的是集群名称为“Flink test”，15个taskManager的内存为8g，然后1个jobManager的内存为8g，yarn上显示总的使用内存资源为128g

这样我们就启动了一个yarn-session 就可以提交flink任务了。
注意我的-d为后台cluster运行，我的此时默认是client模式运行；

2.我们可以使用./bin/flink脚本提交作业，同样我们来看看这个脚本支持哪些参数：

bin/flink
./flink [OPTIONS] [ARGUMENTS]

The following actions are available:
Action “run” compiles and runs a program.
Syntax: run [OPTIONS]
“run” action options:
-c,–class Class with the program entry point(“main” method or “getPlan()” method.Only needed if the JAR file does not specify the class in its manifest.
-C,–classpath Adds a URL to each user code classloader on all nodes in the cluster. The paths must specify a protocol (e.g. file://) and be accessible on all nodes (e.g. by means of a NFS share). You can use this ption multiple times for specifying more than one URL. The protocol must be supported by the {@link java.net.URLClassLoader}.
-d,–detached If present, runs the job in detached mode
-m,–jobmanager host:port Address of the JobManager (master) to which to connect. Specify yarn-cluster’ as the JobManager to deploy a YARN cluster for the job. Use this flag to connect to a different JobManager than the one specified in the configuration.
-p,–parallelism The parallelism with which to run the program. Optional flag to override the default value specified in the configuration.
-q,–sysoutLogging If present, supress logging output to standard out.
-s,–fromSavepoint Path to a savepoint to reset the job back to (for example file:///flink/savepoint-1537).

我们可以使用run选项运行Flink作业。这个脚本可以自动获取到YARN session的地址

线上脚本： nohup bin/flink run -s hdfs:///flink/savepoints/savepoint-bcabee-bf3f54a3b924 -c **** jars/**** test > Flink-RealtimeDAU.log 2>&1 &
或者使用flink的web-ui界面来提交flink应用

通过-s 可以指定savepoints地址，来重跑job。
Savepoint是什么
Flink的savepoint是一个全局的、一致性的快照（snapshot）。其包含两方面：

数据源所有数据的位置；
并行操作的状态
“全局一致”是指所有的输入源数据在指定的位置，所有的并行操作的状态都被完全checkpoint了。
如果你的应用在过去某个时间点做了savepoint，那你随时可以从前面的savepoint更新发布应用。这时，新的应用会从savepoint中的操作状态进行初始化，并从savepoint的数据源位置开始重新处理所有数据。

3.启动之后如何停止运行的程序

关闭jobmanager
线上脚本：bin/flink cancel -s hdfs:///flink/savepoints /savepoints-* -yid application_1535964220442_0034
通过cancel命令进行停止

或者通过yarn application -kill applicationId 直接将yarn-session停止掉

或者通过 flink list 获得 jobId
bin/flink cancel -s hdfs:///flink/savepoints/savepoint-* jobId
其中-s为可选操作

第二种方式

每个job提供一个yarn-session，job运行完成就释放
nohup bin/flink run -m yarn-cluster -yn 15 -s hdfs:///flink/savepoints/111* -c . jars/**** test >./Flink-my-log.log 2>&1 &

其中的-yn是指TaskManager的个数，必须指定。

---

参考：https://blog.csdn.net/asfjgvajfghaklsbf/article/details/82899872