Flink on Yarn 三部曲之三:提交 Flink 任务
全文链接
两种 Flink on YARN 模式
实践之前,对 Flink on YARN 先简单了解一下,如下图所示,Flink on Yarn 在使用的时候分为两种模式,Job Mode 和 Session Mode:
Session Mode:在 YARN 中提前初始化一个 Flink 集群,以后所有 Flink 任务都提交到这个集群,如下图:
Job Mode:每次提交 Flink 任务都会创建一个专用的 Flink 集群,任务完成后资源释放,如下图:
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200407080137105.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLm
NzZG4ubmV0L2JvbGluZ19jYXZhbHJ5,size_16,color_FFFFFF,t_70)
接下来分别实战这两种模式;
准备实战用的数据(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 \
评论