Flink部署及作业提交（On YARN）

Flink部署及作业提交（On YARN）

Hadoop环境快速搭建

官方文档：

YARN Setup

在上一篇 Flink部署及作业提交（On Flink Cluster） 文章中，我们介绍了如何编译部署Flink自身的资源分配和管理系统，并将作业提交到该系统上去运行。但通常来讲这种方式用得不多，因为在企业中，可能会使用不同的分布式计算框架，如Spark、Storm或MapReduce等。

如果每一种框架都需要搭建各自的资源分配和管理系统，就无法共享资源，导致资源利用率低。并且大多企业一般会使用Hadoop生态的相关组件做作为大数据处理平台的底座，如HDFS、Hive、YARN等。

其中 YARN 是资源调度框架、通用的资源管理系统，可以为上层应用提供统一的资源管理和调度，Spark、Flink、Storm等计算框架都可以集成到 YARN 上。如此一来这些计算框架可以享受整体的资源调度，进而提高集群资源的利用率，这也就是所谓的 xxx on YARN。因此，绝大部分企业都是将计算作业放到 YARN 上进行调度，而不是每种计算框架都单独搭一个资源分配和管理系统。这也是为什么要单独介绍Flink On YARN的原因。

想要让Flink作业跑在 YARN 上，我们首先得搭建一个Hadoop环境，为了简单这里只搭建单节点环境。我这里使用的是CDH的Hadoop发行版。下载地址如下：

~]# echo ${JAVA_HOME} /usr/local/jdk/11 [root@hadoop01 ~]# java -version java version "11.0.8" 2020-07-14 LTS Java(TM) SE Runtime Environment 18.9 (build 11.0.8+10-LTS) Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.8+10-LTS, mixed mode) [root@hadoop01 ~]#

配置hosts，将主机名与本地ip建立一个映射关系：

[root@hadoop01 ~]# vim /etc/hosts 192.168.243.142 hadoop01

关闭防火墙：

[root@hadoop01 ~]# systemctl stop firewalld && systemctl disable firewalld

配置免密登录：

[root@hadoop01 ~]# ssh-keygen -t rsa # 生成密钥对 [root@hadoop01 ~]# ssh-copy-id hadoop01 # 拷贝公钥并追加到自己的授权列表文件中

然后就可以开始安装Hadoop了，这里采用 hadoop-2.6.0-cdh5.16.2 版本作为演示，复制下载链接到系统上进行下载：

[root@hadoop01 ~]# cd /usr/local/src [root@hadoop01 /usr/local/src]# wget /usr/local/src]# tar -zxvf hadoop-2.6.0-cdh5.16.2.tar.gz -C /usr/local

配置系统环境变量：

[root@hadoop01 ~]# vim /etc/profile export HADOOP_HOME=/usr/local/hadoop-2.6.0-cdh5.16.2 export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin [root@hadoop01 ~]# source /etc/profile

修改几个配置文件：

[root@hadoop01 ~]# cd $HADOOP_HOME [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/hadoop-env.sh export JAVA_HOME=/usr/local/jdk/11 # 配置JDK的目录 # 配置 core [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/core-site.xml fs.defaultFS hdfs://hadoop01:8020 # 配置 hdfs，设置副本因子和临时目录 [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/hdfs-site.xml dfs.replication 1 hadoop.tmp.dir /data/hadoop/tmp # 配置slave节点的ip或hostname [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/slaves hadoop01 # 配置 yarn [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/yarn-site.xml mapreduce.framework.name yarn # 配置MapReduce [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# cp etc/hadoop/mapred-site.xml.template etc/hadoop/mapred-site.xml [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# vim etc/hadoop/mapred-site.xml yarn.nodemanager.aux-services mapreduce_shuffle # 创建hadoop的临时目录 [root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# mkdir -p /data/hadoop/tmp

应用HDFS的配置：

[root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# ./bin/hdfs namenode -format

启动所有组件：

[root@hadoop01 /usr/local/hadoop-2.6.0-cdh5.16.2]# ./sbin/start-all.sh

启动成功后查看进程：

[root@hadoop01 ~]# jps 3344 SecondaryNameNode 2722 NameNode 3812 Jps 3176 DataNode 3578 NodeManager 3502 ResourceManager [root@hadoop01 ~]#

测试HDFS能否正常读写：

[root@hadoop01 ~]# hadoop fs -put anaconda-ks.cfg / # 任意put一个文件到hdfs [root@hadoop01 ~]# hadoop fs -ls / # 查看hdfs中是否有该文件 Found 1 items -rw-r--r-- 1 root supergroup 1269 2020-09-29 17:45 /anaconda-ks.cfg

经过测试，确认Hadoop环境是运行正常之后，我们就可以尝试将Flink应用放到YARN上运行了。

Flink on YARN两种方式

想要深入了解的话可以参考官方文档：

Deployment Modes

Flink on YARN Session模式实操

首先将在 Flink部署及作业提交（On Flink Cluster） 一文中编译好的Flink目录拷贝到当前部署了Hadoop环境的机器上：

[root@hadoop01 ~]# scp -r 192.168.243.148:/usr/local/src/flink-release-1.11.2/flink-dist/target/flink-1.11.2-bin/flink-1.11.2/ /usr/local/flink

配置环境变量，否则Flink会报找不到Hadoop相关Class的异常：

[root@hadoop01 ~]# vim /etc/profile export HADOOP_MAPRED_HOME=$HADOOP_HOME export HADOOP_COMMON_HOME=$HADOOP_HOME export HADOOP_HDFS_HOME=$HADOOP_HOME export YARN_HOME=$HADOOP_HOME export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop export HADOOP_CLASSPATH=$HADOOP_COMMON_HOME/lib:$HADOOP_HOME/share/hadoop/yarn/*:$HADOOP_HOME/share/hadoop/common/*:$HADOOP_HOME/share/hadoop/mepreduce/*:$HADOOP_HOME/share/hadoop/hdfs/*:$HADOOP_HOME/share/tools/*:$HADOOP_HOME/share/hadoop/--help命令测试一下能否正常输出帮助信息：

[root@hadoop01 ~]# cd /usr/local/flink/ [root@hadoop01 /usr/local/flink]# ./bin/yarn-session.sh --help ... Usage: Optional -at,--applicationType Set a custom application type for the application on YARN -D use value for given property -d,--detached If present, runs the job in detached mode -h,--help Help for the Yarn session CLI. -id,--applicationId Attach to running YARN session -j,--jar Path to Flink jar file -jm,--jobManagerMemory Memory for JobManager Container with optional unit (default: MB) -m,--jobmanager Address of the JobManager to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration. -nl,--nodeLabel Specify YARN node label for the YARN application -nm,--name Set a custom name for the application on YARN -q,--query Display available YARN resources (memory, cores) -qu,--queue Specify YARN queue. -s,--slots Number of slots per TaskManager -t,--ship Ship files in the specified directory (t for transfer) -tm,--taskManagerMemory Memory per TaskManager Container with optional unit (default: MB) -yd,--yarndetached If present, runs the job in detached mode (deprecated; use non-YARN specific option instead) -z,--zookeeperNamespace Namespace to create the Zookeeper sub-paths for high availability mode [root@hadoop01 /usr/local/flink]#

如果没配环境变量的话，执行这条命令就会报找不到类的错误

确认Flink可以正常找到Hadoop后，使用如下命令在 YARN 上创建一个常驻服务：

[root@hadoop01 /usr/local/flink]# ./bin/yarn-session.sh -jm 1024m -tm 2048m ... JobManager Web Interface: # 创建成功的话会输出JobManager的web访问地址

-jm：指定JobManager需要的内存资源 -tm：指定TaskManager需要的内存资源

Tips：要想页面能够正常跳转，还得在浏览器所在主机的hosts文件中配置一下hadoop01这个主机名到IP的映射关系

接下来我们尝试一下提交作业到 YARN 上运行，首先准备好官方提供的测试文件，并put到HDFS中：

[root@hadoop01 ~]# wget -O LICENSE-2.0.txt http://apache.org/licenses/LICENSE-2.0.txt [root@hadoop01 ~]# hadoop fs -copyFromLocal LICENSE-2.0.txt /

然后执行如下命令，提交一个Word Count作业：

[root@hadoop01 ~]# cd /usr/local/flink/ [root@hadoop01 /usr/local/flink]# ./bin/flink run -m hadoop01:37525 ./examples/batch/WordCount.jar \ --input hdfs://hadoop01:8020/LICENSE-2.0.txt --output hdfs://hadoop01:8020/wordcount-result.txt

Tips：这里的hadoop01:37525，是执行完yarn-session.sh命令输出的JobManager的访问地址

执行完成后，控制台会输出如下内容：

Job has been submitted with JobID 2240e11994cf8579a78e16a1984f08db Program execution finished Job with JobID 2240e11994cf8579a78e16a1984f08db has finished. Job Runtime: 10376 ms

除此之外，我们还可以查看该作业输出到HDFS中的结果文件：

[root@hadoop01 /usr/local/flink]# hadoop fs -ls /wordcount-result.txt -rw-r--r-- 1 root supergroup 4499 2020-09-29 20:25 /wordcount-result.txt [root@hadoop01 /usr/local/flink]# hadoop fs -text /wordcount-result.txt

Flink on YARN Per-Job模式实操

首先将之前在 yarn 上运行的应用和相关进程给kill掉：

[root@hadoop01 ~]# yarn application -kill application_1601372571363_0001 [root@hadoop01 ~]# jps 6995 SecondaryNameNode 7204 ResourceManager 7305 NodeManager 11291 Jps 6734 NameNode 6830 DataNode 8942 FlinkYarnSessionCli [root@hadoop01 ~]# kill 8942

Per-Job模式更简单，因为是提交一个作业就创建一次资源的，所以直接运行如下命令就可以提交一个Flink的Word Count作业到 yarn 上，不需要像Session模式那样事先去创建资源：

[root@hadoop01 /usr/local/flink]# ./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar

Flink Scala Shell的简单使用

在之前的演示中可以看到，提交的Flink作业都是以jar包形式存在的。如果我们在实际开发中，需要频繁修改代码提交到 yarn 上测试，那么就得频繁的打包，相对来说就有点麻烦。那么Flink有没有像Spark那样提供类似于 Spark Shell 的交互式编程终端用于简单的代码测试呢？答案是有的，Flink提供了PyFlink Shell和Scala Shell，可以执行Python和Scala代码。

这里简单演示下Flink Scala Shell的使用，执行如下命令打开Flink Scala Shell：

[root@hadoop01 /usr/local/flink]# ./bin/start-scala-shell.sh local

这里的local表示在本地运行，除此之外还可以选择remote和yarn，具体可以使用--help参数进行查看

shell里调用API的方式还是一样的，只是环境变成了内置的变量，例如这里使用的benv就表示批处理的env：

scala> val dataSet = benv.readTextFile("file:///root/LICENSE-2.0.txt") dataSet: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@3110bb19 scala> dataSet.print

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。