2 Flink 部署及启动

Flink 支持多种安装模式:
1.local（本地）——单机模式，一般不使用；
2.standalone——独立模式，Flink 自带集群，开发测试环境使用；
3.yarn——计算资源统一由 Hadoop YARN 管理，生产环境使用。
Flink 集群的安装不属于本文档的范畴，如安装 Flink，可自行搜索资料进行安装。
本节重点在 Flink 的 Yarn 部署模式。
在一个企业中，为了最大化的利用集群资源，一般都会在一个集群中同时运行多种类型的 Workload，可以使用 YARN 来管理所有计算资源。

1. Flink 在 Yarn 上的部署架构

从图中可以看出，Yarn 的客户端需要获取 hadoop 的配置信息，连接 Yarn 的 ResourceManager。所以要设置 YARN_CONF_DIR 或者 HADOOP_CONF_DIR 或者 HADOOP_CONF_PATH，只要设置了其中一个环境变量，就会被读取。如果读取上述的变量失败了，那么将会选择 hadoop_home 的环境变量，会尝试加载$HADOOP_HOME/etc/hadoop 的配置文件。
1.当启动一个 Flink Yarn 会话时，客户端首先会检查本次请求的资源（存储、计算）是否足够。资源足够将会上传包含 HDFS 及 Flink 的配置信息和 Flink 的 jar 包到 HDFS；
2.客户端向 RM 发起请求；
3.RM 向 NM 发请求指令，创建 container，并从 HDFS 中下载 jar 以及配置文件；
4.启动 ApplicationMaster 和 jobmanager,将 jobmanager 的地址信息写到配置文件中，再发到 hdfs 上；
5.同时，AM 向 RM 发送心跳注册自己，申请资源（cpu、内存）；
6.创建 TaskManager 容器，从 HDFS 中下载 jar 包及配置文件并启动；
7.各 task 任务通过 jobmanager 汇报自己的状态和进度，AM 和 jobmanager 在一个容器上，AM 就能掌握各任务的运行状态，从而可以在任务失败时，重新启动任务；
8.任务完成后，AM 向 RM 注销并关闭自己；

2. 启动集群

1.修改 hadoop 的配置参数： vim etc/hadoop/yarn-site.xml
添加：

<property>
         <name>yarn.nodemanager.vmem-check-enabled</name>
         <value>false</value>
</property>

修改 Hadoop 的 yarn-site.xml，添加该配置表示内存超过分配值，是否将任务杀掉。
默认为 true。 运行 Flink 程序，很容易内存超标，这个时候 yarn 会自动杀掉 job。
2.修改全局变量 /etc/profile：
添加：export HADOOP_CONF_DIR=/export/servers/hadoop/etc/Hadoop
YARN_CONF_DIR 或者 HADOOP_CONF_DIR 必须将环境变量设置为读取 YARN 和 HDFS 配置
2.启动 HDFS、zookeeper（如果是外置 zookeeper）、YARN 集群；
3.使用 yarn-session 的模式提交作业。
Yarn Session 模式提交作业有两种方式：yarn-session 和 yarn-cluster

3. 模式一: yarn-session

特点：
1.使用 Flink 中的 yarn-session（yarn 客户端），会启动两个必要服务 JobManager 和 TaskManagers；
2.客户端通过 yarn-session 提交作业；
3.yarn-session 会一直启动，不停地接收客户端提交的任务；
4.如果拥有有大量的小作业，适合使用这种方式。

在 flink 目录启动 yarn-session：
bin/yarn-session.sh -n 2 -tm 800 -jm 800 -s 1 -d
-n 表示申请 2 个容器
-s 表示每个容器启动多少个 slot 离模式，表示以后台程
-tm 表示每个 TaskManager 申请 800M 内存
-d 分序方式运行
使用 flink 提交任务：
bin/flink run examples/batch/WordCount.jar
如果程序运行完了，可以使用 yarn application -kill application_id 杀掉任务：
yarn application -kill application_1554377097889_0002
bin/yarn-session.sh -n 2 -tm 800 -s 1 -d 意思是:
同时向 Yarn 申请 3 个 container（即便只申请了两个，因为 ApplicationMaster 和 Job Manager 有一个额外的容器。一旦将 Flink 部署到 YARN 群集中，它就会显示 Job Manager 的连接详细信息），其中 2 个 Container 启动 TaskManager（-n 2），每个 TaskManager 拥有两个 Task Slot（-s 1），并且向每个 TaskManager 的 Container 申请 800M 的内存，以及一个 ApplicationMaster（Job Manager）。

4. 模式二: yarn-cluster

特点：
1.直接提交任务给 YARN；
2.大作业，适合使用这种方式；
3.会自动关闭 session。

使用 flink 直接提交任务：
bin/flink run -m yarn-cluster -yn 2 -yjm 800 -ytm 800 /export/servers/flink-1.6.0/examples/batch/WordCount.jar
-yn 表示 TaskManager 的个数
注意：
1.在创建集群的时候，集群的配置参数就写好了，但是往往因为业务需要，要更改一些配置参数，这个时候可以不必因为一个实例的提交而修改 conf/flink-conf.yaml;
可以通过：-D Dynamic properties 来覆盖原有的配置信息：比如：
-Dfs.overwrite-files=true -Dtaskmanager.network.numberOfBuffers=16368
2.如果使用的是 flink on yarn 方式，想切换回 standalone 模式的话，需要删除：/tmp/.yarn-properties-root，因为默认查找当前 yarn 集群中已有的 yarn-session 信息中的 jobmanager。