单元10 任务10.3部署运行Flink on YARN

单元10Flink流式计算框架部署与操作任务10.3部署运行Flinkon

YARN任务10.3部署运行FlinkonYARN【任务场景】小张：经理，Spark独立运行模式的集群已经部署好，可以正常使用了。经理：Flink是继Storm和Spark之后的第三代流引擎框架，支持在有界和无界数据流上做有状态计算，以事件为单位，并且支持SQL、State、WaterMark等。它支持"exactlyonce"，即事件投递保证只有一次，数据的准确性能得到提升。相比Storm，它吞吐量更高，延迟更低；相比SparkStreaming，Flink是真正意义上的实时计算，且所需计算资源相对更少。但是咱们部署的Flink独立模式能处理的数据量有限，主流的趋势还是Flink与Hadoop进行集成开发应用。小张：好的，那我修改配置，基于Hadoop的yarn资源管理器来支持Flink的运行。经理：好的，尽快掌握Flinkonyarn的运行模式。小张：好的，没问题。任务10.3部署运行FlinkonYARN【任务布置】本任务要求完成Flinkonyarn模式的部署和运行；基于yarn模式执行词频统计程序，将统计结果写入指定的文件；通过yarn的Web端口监控Flink任务的执行状态。知识点1：FlinkonYARN的运行方法FlinkonYARN部署很简单，就是只要部署好hadoop集群即可，我们只需要部署一个Flink客户端，然后从flink客户端提交Flink任务即可。Flinkonyarn有两种运行方式：1.方式一第一种方式是在yarn上启动一个Flink集群，flink就持有了yarn的资源（即使flink上没有任何程序，yarn的资源其他非flink集群任务使用不了了，造成资源浪费），在flink集群上提交任务。除非把Flink集群停了，不然资源不会释放。也就是【yarn-session.sh(开辟资源)+flinkrun(提交任务)】的运行方式。yarn-session命令的用法如下表所示：知识点1：FlinkonYARN的运行方法选择

命令行参数含义必选-n,--container<arg>分配多少个yarn容器(=taskmanager的数量)可选-D<arg>动态属性

可选-d,--detached-d,--detached可选-jm,--jobManagerMemory<arg>JobManager的内存[inMB]可选-nm,--name在YARN上为一个自定义的应用设置一个名字

可选-q,--query显示yarn中可用的资源(内存,cpu核数)可选-qu,--queue<arg>指定YARN队列.可选-s,--slots<arg>每个TaskManager使用的slots数量

可选-tm,--taskManagerMemory<arg>每个TaskManager的内存[inMB]可选-z,--zookeeperNamespace<arg>针对HA模式在zookeeper上创建NameSpace可选-id,--applicationId<yarnAppId>YARN集群上的任务id，附着到一个后台运行的yarnsession中知识点1：FlinkonYARN的运行方法【运行示例1】启动一个一直运行的flink集群/bin/yarn-session.sh-n2-jm1024-tm1024[-d]把任务附着到一个已存在的flinkyarnsession./bin/yarn-session.sh-idapplication_1463870264508_0029执行任务./bin/flinkrun./examples/batch/WordCount.jar-inputhdfs://hadoop100:9000/LICENSE-outputhdfs://hadoop100:9000/wordcount-result.txt如果需要停止任务可以在web界面停止或者命令行执行cancel命令。知识点1：FlinkonYARN的运行方法2.方式二第二种方式是每提交一个任务就在yarn上面启动一个flink小集群，通常推荐使用这种方法运行，任务运行完了资源会自动释放。主要使用的命令是【flinkrun-myarn-cluster(开辟资源+提交任务)】，flinkrun命令的常用用法如下表所示：选择

命令行参数含义必选-n,--container<arg>分配多少个yarn容器(=taskmanager的数量)可选-D<arg>动态属性可选-d,--detached独立运行可选-jm,--jobManagerMemory<arg>JobManager的内存[inMB]可选-nm,--name在YARN上为一个自定义的应用设置一个名字可选-q,--query显示yarn中可用的资源(内存,cpu核数)可选-qu,--queue<arg>指定YARN队列.可选-s,--slots<arg>每个TaskManager使用的slots数量可选-tm,--taskManagerMemory<arg>每个TaskManager的内存[inMB]可选-z,--zookeeperNamespace<arg>针对HA模式在zookeeper上创建NameSpace可选-id,--applicationId<yarnAppId>YARN集群上的任务id，附着到一个后台运行的yarnsession中知识点1：FlinkonYARN的运行方法flinkrun[OPTIONS]<jar-file><arguments>命令中run的操作参数主要包括以下3个：命令参数含义-c,--class<classname>如果没有在jar包中指定入口类，则需要在这里通过这个参数指定-m,--jobmanager<host:port>指定需要连接的jobmanager(主节点)地址，使用这个参数可以指定一个不同于配置文件中的jobmanager

-p,--parallelism<parallelism>指定程序的并行度。可以覆盖配置文件中的默认值。知识点1：FlinkonYARN的运行方法【案例1】默认查找当前yarn集群中已有的yarn-session信息中的jobmanager【/tmp/.yarn-properties-root】：./bin/flinkrun./examples/batch/WordCount.jar-inputhdfs://hostname:port/hello.txt-outputhdfs://hostname:port/result1【案例2】连接指定host和port的jobmanager：./bin/flinkrun-mhadoop100:1234./examples/batch/WordCount.jar-inputhdfs://hostname:port/hello.txt-outputhdfs://hostname:port/result1【案例3】启动一个新的yarn-session：./bin/flinkrun-myarn-cluster-yn2./examples/batch/WordCount.jar-inputhdfs://hostname:port/hello.txt-outputhdfs://hostname:port/result1【小提示】yarnsession命令行的选项也可以使用./bin/flink工具获得。它们都有一个y或者yarn的前缀。例如：./bin/flinkrun-myarn-cluster-yn2./examples/batch/WordCount.jar知识点2：故障调试与恢复1.Flinkonyarn的故障恢复Flink的yarn客户端通过下面的配置参数来控制容器的故障恢复。这些参数可以通过conf/flink-conf.yaml或者在启动yarnsession的时候通过-D参数来指定。yarn.reallocate-failed：这个参数控制了flink是否应该重新分配失败的taskmanager容器。默认是true。yarn.maximum-failed-containers：applicationMaster可以接受的容器最大失败次数，达到这个参数，就会认为yarnsession失败。默认这个次数和初始化请求的taskmanager数量相等(-n参数指定的)。yarn.application-attempts：applicationMaster重试的次数。如果这个值被设置为1(默认就是1)，当applicationmaster失败的时候，yarnsession也会失败。设置一个比较大的值的话，yarn会尝试重启applicationMaster。知识点2：故障调试与恢复2.调试失败的yarnsession一个Flinkyarnsession部署失败可能会有很多原因。一个错误的hadoop配置(hdfs权限，yarn配置)，版本不兼容(使用cdh中的hadoop运行flink)，或者其他的错误。flinkyarnsession部署失败是由于它自身的原因，用户必须依赖于yarn的日志来进行分析。

最有用的就是yarnlogaggregation，用户必须在yarn-site.xml文件中设置yarn.log-aggregation-enable属性为true，才能启动此功能。启用yarnlogaggregation之后，用户可以通过下面的命令来查看一个失败的yarnsession的所有详细日志：yarnlogs-applicationId<applicationID>。如果错误发生在运行时(例如某个taskmanager停止工作了一段时间)，flinkyarnclient也会打印一些错误信息在控制台上。除此之外，yarnresourcemanager的web界面(默认端口是8088)，也可以用于在yarn程序运行期间或运行失败时查看日志定位问题。FlinkonYARN部署流程【工作流程】

部署yarn运行模式的Flink基本工作流程包括：1.Flink集群部署规划；2.修改yarn配置文件；3.以Flinkonyarn模式提交运行词频统计样例程序；4.通过yarn的WebUI监控Flink任务执行情况。FlinkonYARN部署流程1.集群规划

Flinkonyarn模式需要依赖Hadoop集群，3个节点的Flink集群规划如下表所示：主机名节点环境用途master01CentOS7、JDK1.8、Hadoop3.1.1、Flink客户端主节点worker01CentOS7、JDK1.8、Hadop3.1.1从节点1worker02CentOS7、JDK1.8、Hadoop3.1.1从节点2FlinkonYARN部署流程2.修改yarn配置文件修改Hadoop集群的yarn-site.xml配置文件，每个文件中增加以下配置内容：

<property> <name>yarn.nodemanger.pmem-check-enabled</name> <value>false</value> </property> <property> <name>yarn.nodemanger.vmem-check-enabled</name> <value>false</value> </property>FlinkonYARN部署流程3.以Flinkonyarn模式提交运行词频统计样例程序在终端通过flinkrun命令提交执行词频统计样例程序，命令行中通过--input选项指定统计的源文件为/root/anaconda-ks.cfg，执行结果通过--output指定/root/out1，执行命令及过程如下所示：[root@master01bin]#flinkrun-myarn-cluster../examples/streaming/WordCount.jar--input/root/anaconda-ks.cfg--output/root/out1SLF4J:ClasspathcontainsmultipleSLF4Jbindings.SLF4J:Foundbindingin[jar:file:/opt/flink/lib/log4j-slf4j-impl-2.12.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J:Foundbindingin[jar:file:/opt/hado