这将是一套完整详细且持续更新的、长期维护的 原创 Flink系列教程、文档，其中会包含各种商用实例详解、Flink源码讲解、机制剖析、周边组件讲解等，旨在帮助开发者快速融入Flink开发，或作为工具文档参阅。也希望更多的开发者可以参与到大数据相关的技术讨论中来，共同商讨疑难杂症，丰富社区。——尽际
本文标题: 基于 YARN 模式部署 Flink 服务
文章作者: 尽际
发布时间: 2019年07月28日 - 23:02:30
最后更新: 2019年07月28日 - 23:02:30
原始链接:
许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。
三、 基于 YARN 模式部署 Flink 服务
3.1 基于 YARN 运行 Flink 服务的原理
在 YARN 上启动 Flink 服务的过程如图3-1 所示（Flink 官方图示）。
-
图3-1 在 YARN 上启动 Flink 服务的过程YARN 上启动 Flink 服务的过程.png
3.1.1 Flink-Session 方式
下面将对图3-1 所示内容进行阐述。
-
准备资源
-
发现配置
想要让 Flink 服务运行与 YARN 之上，首先需要让 Flink 能够发现 YARN 和 HDFS 的相关配置，因此，需要通过 YARN_CONF_DIR、HADOOP_CONF_DIR、HADOOP_CONF_PATH 属性来指定 Hadoop 配置文件所在目录。
提示：上述 3 个属性，指定其中之一即可，如果全部指定，将按照指定顺序加载。习惯上，常用 HADOOP_CONF_DIR 属性指定。 -
检查可用物理资源
检查 YARN 集群中的可用内存和虚拟机CPU内核是否充足，如果充足，则进行下一步。
-
上传资源。
然后，启动 Flink 服务时，会根据 Hadoop 配置，将 Flink 服务依赖的 Jar 包和配置文件上传至 HDFS。
-
-
装配Application Master容器
此时为了开启 YARN 的 Application Master 服务，需要一个准备好的 Container 实例。这个过程会伴随如下步骤。
-
注册 Container 所需资源
将上一步上传的配置文件和 jar 包（统称为Container外部资源）注册为即将创建的 Container 内部的资源。这个过程其实是将 外部资源的存放路径（HDFS上）赋值给 Container 内部的变量。
-
分配 NodeManager 并实例化 Container
根据 Container 中注册（存储）的外部资源的路径，下载对应配置文件和 Jar 包到 Container 所在的机器节点，然后完成该 Container 的实例化。
-
启动 Application Master
该 Container 实例化完成后（在某个NodeManager上），即可启动 Application Master 服务。
-
-
启动 JobManager 服务
Application Master 服务启动成功后，会启动 Flink 的 JobManager 服务，它将与 ApplicationMaster 服务在同一个 Container 上运行，因此与 ApplicationMaster 通信的客户端（YarnClusterClient）就可以将 JobManager的地址（与ApplicationMaster相同）暴露返回给客户端。从而在后续提交 Job 的过程中，可以知道向哪个 JobManager 提交 Flink Job。
提示：由于ApplicationMaster 和 JobManager 是随机（某种程度上）在某个 NodeManager 的 Container 上运行的，因此，一个 YARN 集群可以运行多个 ApplicationMaster 和 JobManager 实例。至此，已经完成了在 YARN 上开启一个 Flink Session 的过程。
-
提交 Flink Job
在向 JobManager 提交 Flink Job 时，才会分配 TaskManager 所需的 Container（容器），具体步骤如下。
-
提交任务
找到某个已经开启的 Flink Session，并找到该 Session 下的 JobManager 通信地址，然后发起提交Flink Job 请求。
-
分配 Container 开启 TaskManager
根据 Flink Job 中配置的相关参数，开启指定个数，指定内存占用的 TaskManager，并开始执行任务，执行完毕后，回收该 TaskManager 及 Containner。注意，此时并不会回收 JobManager，因而，在这种方式下，同一个 Flink Session 可以用于执行多个 Flink Job。
至此，在 YARN 上启动 Flink Session 并提交运行 Flink Job 的过程就结束了。
-
3.1.2 Flink Single Job 方式
Flink Single Job 方式，也称之为 Flink Per Job 方式。这种方式与 Flink Session 方式大体相同，不同的是Single Job 方式不必提前在 YARN 上开启 JobManager，而是每次提交 Flink Job 的时候，动态创建 JobManager（也在Containner中），然后启动 TaskManager（Container中） 执行任务。然后再Flink Job 结束后，销毁释放本次 Job 所使用的 JobManager 和 TaskManager。可以理解为是一次性的 Flink Session。
了解完上述内容之后，我们将基于两种方式在 YARN 上创建 Flink 服务并提交 Flink Job。
3.2 基于 YARN 模式部署 Flink 服务（Flink-Session 方式）
了解了 Flink on YARN 的基本概念后，开始根据如下步骤部署服务吧。
在本例中，YARN 服务分布情况见表3-1。
-
表3-1 YARN 服务分布情况
linux01 linux02 linux03 - Resourcemanager - NodeManager NodeManager NodeManager 提示：需要开启 YARN 高可用开发者可参考笔者之前的文章自行配置。此时开发者必须确保 YARN 服务已经正常运行，并且开启了 Hadoop 的 JobHistoryServer 服务
-
解压 Flink 压缩包
将Flink压缩包上传至第一台虚拟机的 ~/softwares/installations 目录后，使用如下命令进行解压：
$ tar -zxf tar -zxf ~/softwares/installations/flink-1.8.1-bin-scala_2.11.tgz -C ~/modules/ -
为当前会话设置 Hadoop 环境变量
当前会话是指通过 ssh 访问 Linux的客户端，比如 xShell 中的某一个窗口，或者 CRT 中的某个窗口。
想要让 Flink 服务运行与 YARN 之上，首选，需要让 Flink 能够发现 Hadoop 的相关配置，操作命令如下：
$ pwd /home/admin/modules/flink-1.8.1 $ export HADOOP_CONF_DIR=/home/admin/modules/hadoop-2.7.2/etc/hadoop $ export HADOOP_HOME=/home/admin/modules/hadoop-2.7.2 $ export HADOOP_CLASSPATH=`/home/admin/modules/hadoop-2.7.2/bin/hadoop classpath`完成上述命令后，在当前会话中可以通过如下命令查看对应的变量是否被正确赋值：
$ echo $HADOOP_CLASSPATH 输出如下： /home/admin/modules/hadoop-2.7.2/etc/hadoop:/home/admin/modules/hadoop-2.7.2/share/hadoop/common/lib/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/common/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/hdfs:/home/admin/modules/hadoop-2.7.2/share/hadoop/hdfs/lib/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/hdfs/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/yarn/lib/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/yarn/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/mapreduce/lib/*:/home/admin/modules/hadoop-2.7.2/share/hadoop/mapreduce/*:/home/admin/modules/hadoop-2.7.2/contrib/capacity-scheduler/*.jar输出上述内容后，即证明为当前会话成功配置了 Hadoop 环境变量。
提示：在生产环境中，也多用这种临时导入Hadoop环境变量的方式进行配置，即每次新的 ssh 会话中操作Flink on YARN都会提前 export Hadoop相关的环境变量。开发者也可选择将上述命令封装到某个脚本中，以方便后续操作。 -
合理配置依赖包
-
添加 Flink-Table 依赖包
在生产环境中，开发者开发的 Flink 应用程序可能会伴随 Table API 的操作（支持以 SQL 的方式进行数据分析），因此，开发者需要将 Flink 安装目录下的 opt 中的 flink-table_2.11-1.8.1.jar 包拷贝到 lib 目录中。操作命令如下所示：
$ pwd /home/admin/modules/flink-1.8.1 $ cp -a opt/flink-table_2.11-1.8.1.jar lib/ -
添加 Hadoop 依赖包
将之前下载好的 flink-shaded-hadoop-2-uber-2.7.5-7.0.jar 包移动至 Flink 安装目录的 lib 目录中，使 Flink 能够支持对 Hadoop 的操作。操作命令如下所示：
$ pwd /home/admin/modules/flink-1.8.1 $ cp -a ~/softwares/installations/flink-shaded-hadoop-2-uber-2.7.5-7.0.jar ./lib/
以上操作完成后，即可看到 lib 目录下的内容如下所示：
$ ll lib/ flink-dist_2.11-1.8.1.jar flink-shaded-hadoop-2-uber-2.7.5-7.0.jar flink-table_2.11-1.8.1.jar log4j-1.2.17.jar slf4j-log4j12-1.7.15.jar -
提示：Flink on YARN模式中，不需要每台节点都安装 Flink，只需要在操作的节点上，安装 Flink 即可。
-
在 YARN 上启动 Flink-Session
在 YARN 上运行一个 Flink-Session，操作命令如下：
${FLINK_HOME}/bin/yarn-session.sh […]例如：
$ pwd /home/admin/modules/flink-1.8.1 $ bin/yarn-session.sh -n 2 -jm 512m -tm 1024m上述命令执行完毕后，即可在 YARN 的 8088 端口（Resourcemanager 后台管理页面）看到有一个 YARN 任务处于正在运行的状态，如图3-2 所示。
-
图3-2 Resourcemanager 后台管理页面Resourcemanager 后台管理页面.png
-
图3-2 Resourcemanager 后台管理页面
此时，单击 ApplicationMaser 按钮，即可进入 Flink 的后台管理页面，如图3-3 所示。
-
图3-3 Flink 后台管理页面Flink 后台管理页面.png
提示：由于现在没有运行任何 Flink Job，因此，该Flink-Session 中还没有分配 Container 来运行 TaskManager。因此，在上图中看到的 TaskManager 数量为 0。
如果此时需要关闭该 Flink-Session，在控制台输入 stop 即可。
但是，如果在启动Flink-Session时，通过添加 -d 参数，开启了 detached，则可以通过 YARN 的 kill 命令，来关闭该 Flink-Session，命令如下：
$ {HADOOP_HOME}/bin/yarn application -kill <application_id>
-
提交 Flink Job 到指定的 Flink-Session中
首先，查看刚才启动 Flink-Session 时，控制台返回的 JobManager 通信地址为 linux01:42535（或记录当前Flink-Session 的 applicaitonID为 application_1564330446958_0001），如图3-4 所示。
-
图3-4 Flink-Session 中 JobManager 的通信地址Flink-Session地址.png
-
图3-4 Flink-Session 中 JobManager 的通信地址
然后，开启一个新的 ssh 会话，重新导入 Hadoop 环境变量，命令如下：
$ export HADOOP_CONF_DIR=/home/admin/modules/hadoop-2.7.2/etc/hadoop
$ export HADOOP_HOME=/home/admin/modules/hadoop-2.7.2
$ export HADOOP_CLASSPATH=`~/modules/hadoop-2.7.2/bin/hadoop classpath`
接着，提交 Flink Job 到指定的 Flink-Session上运行，命令如下（-m 参数的方式）：
$ pwd
/home/admin/modules/flink-1.8.1
$ bin/flink run \
-m linux01:42535 \
-p 1 \
./examples/batch/WordCount.jar
或者，使用命令如下（-yid 参数的方式）：
$ pwd
/home/admin/modules/flink-1.8.1
$ bin/flink run \
-yid application_1564330446958_0001 \
-p 1 \
./examples/batch/WordCount.jar
提示：如果在提交 Flink Job 到 Flink-Session 的过程中，没有通过 -m 或 -yid 参数指定目标 Session，则默认将Flink Job 提交到最后一个开启的 Flink-Session 上。
上述两条命令分别运行测试后，即可在 Job Manager的后台管理页面看到如图3-5 所示内容。
-
图3-5 Flink Job 在 Session 上运行成功Flink Job 运行成功.png
3.3 基于 YARN 模式部署 Flink 服务（Flink-Single[Per] Job 方式）
-
前置准备
这部分的准备工作与3.2 小节中的 1~3 的操作内容一致，不再赘述。
-
在 YARN 上启动并提交 Flink-Single Job
前文提到过，Single Job这种方式为一次性执行，即，当前 Flink Job 执行完毕，立即释放销毁 JobManager 和 TaskManager。因此，它的操作过程将非常简单。这种模式一般多用于执行长期运行的流式处理作业（Flink Streaming Job）。
在这种方式下，开启服务并提交 Flink Job 的命令如下：
$ export HADOOP_CONF_DIR=/home/admin/modules/hadoop-2.7.2/etc/hadoop $ export HADOOP_HOME=/home/admin/modules/hadoop-2.7.2 $ export HADOOP_CLASSPATH=`~/modules/hadoop-2.7.2/bin/hadoop classpath` $ bin/flink run \ -m yarn-cluster \ -p 10 \ -yn 5 \ ./examples/batch/WordCount.jar提示：Single Job 这种方式下，-m 参数的固定值为 yarn-cluster，-p 参数为任务并行度，-yn 为开启的TaskManager 个数。运行后，查看 Job 状态的方式与 3.2 小节相同。
3.4 相关重要参数说明
3.4.1 开启 Flink-Session 的重要参数
重要参数见表3-2 。
-
表3-2 开启 Flink-Session 的重要参数
参数名 含义 -d 开启detached模式。 -jm JobManager的内存，单位：MB。 -nm 该Flink Cluster在YARN上对应的application名称。 -s 每一个TaskManager上的slots数量。 -tm 每一个TaskManager的内存，单位：MB。 -nm 当前Application的名称。
3.4.2 向 Flink-Session 提交 Flink Job 的重要参数
重要参数见表3-3。
-
表3-3 向 Flink-Session 提交 Flink Job 的重要参数
参数名 含义 -m JobManager地址。-m 与 -yid 二选一。 -yid YARN 的 ApplicationId。-m 与 -yid 二选一。 关于提交 Flink Job 的其余重要参数见 3.4.3 小结。
3.4.3 提交 Single Job 的重要参数
重要参数见表3-4。
-
表3-4 提交 Single Job 的重要参数
参数名 含义 -m 固定为yarn-cluster -yj Flink jar的所在目录。一般为Flink根目录下的lib目录。 -yjm 指定JobManager所在的Container内存。单位：MB。 -yn 为TaskManager数量。 -ynm YARN中application的名称。 -ys 每一个TaskManager中slots的数量。 -ytm 每一个TaskManager Container的内存，单位MB。 -c 指定Job对应的jar包中主函数所在类名。
