Flink on Yarn 三部曲之三：提交 Flink 任务

作者：程序员欣宸

2022-12-04
广东
本文字数：1918 字
阅读完需：约 6 分钟

欢迎访问我的 GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

本文是《Flink on Yarn 三部曲》系列的终篇，先简单回顾前面的内容：

《Flink on Yarn 三部曲之一：准备工作》：准备好机器、脚本、安装包；
《Flink on Yarn 三部曲之二：部署和设置》：完成 CDH 和 Flink 部署，并在管理页面做好相关的设置；

现在 Flink、Yarn、HDFS 都就绪了，接下来实践提交 Flink 任务到 Yarn 执行；

全文链接

两种 Flink on YARN 模式

实践之前，对 Flink on YARN 先简单了解一下，如下图所示，Flink on Yarn 在使用的时候分为两种模式， Job Mode 和 Session Mode ：
Session Mode ：在 YARN 中提前初始化一个 Flink 集群，以后所有 Flink 任务都提交到这个集群，如下图：
Job Mode ：每次提交 Flink 任务都会创建一个专用的 Flink 集群，任务完成后资源释放，如下图：
接下来分别实战这两种模式；

准备实战用的数据(CDH 服务器)

接下来提交的 Flink 任务是经典的 WordCount，先在 HDFS 中准备一份文本文件，后面提交的 Flink 任务都会读取这个文件，统计里面每个单词的数字，准备文本的步骤如下：
SSH 登录 CDH 服务器；
切换到 hdfs 账号： su - hdfs
下载实战用的 txt 文件：

wget https://github.com/zq2599/blog_demos/blob/master/files/GoneWiththeWind.txt

复制代码

创建 hdfs 文件夹： hdfs dfs -mkdir /input
将文本文件上传到/input 目录： hdfs dfs -put ./GoneWiththeWind.txt /input
准备工作完成，可以提交任务试试了。

Session Mode 实战

SSH 登录 CDH 服务器；
切换到 hdfs 账号： su - hdfs
进入目录： /opt/flink-1.7.2/
执行如下命令创建 Flink 集群， -n 参数表示 TaskManager 的数量， -jm 表示 JobManager 的内存大小， -tm 表示每个 TaskManager 的内存大小：

./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024

复制代码

创建成功后，控制台输出如下图，注意红框中的提示，表明可以通过 38301 端口访问 Flink：
浏览器访问 CDH 服务器的 38301 端口，可见 Flink 服务已经启动：
浏览器访问 CDH 服务器的 8088 端口，可见 YARN 的 Application(即 Flink 集群)创建成功，如下图，红框中是任务 ID，稍后结束 Application 的时候会用到此 ID：
再开启一个终端，SSH 登录 CDH 服务器，切换到 hdfs 账号，进入目录： /opt/flink-1.7.2
执行以下命令，就会提交一个 Flink 任务（安装包自带的 WordCount 例子），并指明将结果输出到 HDFS 的 wordcount-result.txt 文件中：

bin/flink run ./examples/batch/WordCount.jar \-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \-output hdfs://192.168.50.134:8020/wordcount-result.txt

复制代码

执行完毕后，控制台输出如下：
flink 的 WordCount 任务结果保存在 hdfs，我们将结果取出来看看： hdfs dfs -get /wordcount-result.txt
vi 打开 wordcount-result.txt 文件，如下图，可见任务执行成功，指定文本中的每个单词数量都统计出来了：
浏览器访问 Flink 页面(CDH 服务器的 38301 端口)，也能看到任务的详细情况：
销毁这个 Flink 集群的方法是在控制台执行命令： yarn application -kill application_1580173588985_0002
Session Mode 的实战就完成了，接下来我们来尝试 Job Mode；

Job Mode

执行以下命令，创建一个 Flink 集群，该集群只用于执行参数中指定的任务(wordCount.jar)，结果输出到 hdfs 的 wordcount-result-1.txt 文件：

bin/flink run -m yarn-cluster \-yn 2 \-yjm 1024 \-ytm 1024 \./examples/batch/WordCount.jar \-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \-output hdfs://192.168.50.134:8020/wordcount-result-1.txt

复制代码

控制台输出如下，表明任务执行完成：
如果您的内存和 CPU 核数充裕，可以立即执行以下命令再创建一个 Flink 集群，该集群只用于执行参数中指定的任务(wordCount.jar)，结果输出到 hdfs 的 wordcount-result-2.txt 文件：

bin/flink run -m yarn-cluster \-yn 2 \-yjm 1024 \-ytm 1024 \./examples/batch/WordCount.jar \-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \-output hdfs://192.168.50.134:8020/wordcount-result-2.txt

复制代码

在 YARN 管理页面可见任务已经结束：
执行命令 hdfs dfs -ls / 查看结果文件，已经成功生成：
执行命令 hdfs dfs -get /wordcount-result-1.txt 下载结果文件到本地，检查数据正常；
至此，Flink on Yarn 的部署、设置、提交都实践完成，《Flink on Yarn 三部曲》系列也结束了，如果您也在学习 Flink，希望本文能够给您一些参考，也建议您根据自身情况和需求，修改 ansible 脚本，搭建更适合自己的环境；