單元10 任務(wù)10.3部署運(yùn)行Flink on YARN

單元10Flink流式計(jì)算框架部署與操作任務(wù)10.3部署運(yùn)行Flinkon

YARN任務(wù)10.3部署運(yùn)行FlinkonYARN【任務(wù)場(chǎng)景】小張：經(jīng)理，Spark獨(dú)立運(yùn)行模式的集群已經(jīng)部署好，可以正常使用了。經(jīng)理：Flink是繼Storm和Spark之后的第三代流引擎框架，支持在有界和無(wú)界數(shù)據(jù)流上做有狀態(tài)計(jì)算，以事件為單位，并且支持SQL、State、WaterMark等。它支持"exactlyonce"，即事件投遞保證只有一次，數(shù)據(jù)的準(zhǔn)確性能得到提升。相比Storm，它吞吐量更高，延遲更低；相比SparkStreaming，F(xiàn)link是真正意義上的實(shí)時(shí)計(jì)算，且所需計(jì)算資源相對(duì)更少。但是咱們部署的Flink獨(dú)立模式能處理的數(shù)據(jù)量有限，主流的趨勢(shì)還是Flink與Hadoop進(jìn)行集成開(kāi)發(fā)應(yīng)用。小張：好的，那我修改配置，基于Hadoop的yarn資源管理器來(lái)支持Flink的運(yùn)行。經(jīng)理：好的，盡快掌握Flinkonyarn的運(yùn)行模式。小張：好的，沒(méi)問(wèn)題。任務(wù)10.3部署運(yùn)行FlinkonYARN【任務(wù)布置】本任務(wù)要求完成Flinkonyarn模式的部署和運(yùn)行；基于yarn模式執(zhí)行詞頻統(tǒng)計(jì)程序，將統(tǒng)計(jì)結(jié)果寫(xiě)入指定的文件；通過(guò)yarn的Web端口監(jiān)控Flink任務(wù)的執(zhí)行狀態(tài)。知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法FlinkonYARN部署很簡(jiǎn)單，就是只要部署好hadoop集群即可，我們只需要部署一個(gè)Flink客戶端，然后從flink客戶端提交Flink任務(wù)即可。Flinkonyarn有兩種運(yùn)行方式：1.方式一第一種方式是在yarn上啟動(dòng)一個(gè)Flink集群，flink就持有了yarn的資源（即使flink上沒(méi)有任何程序，yarn的資源其他非flink集群任務(wù)使用不了了，造成資源浪費(fèi)），在flink集群上提交任務(wù)。除非把Flink集群停了，不然資源不會(huì)釋放。也就是【yarn-session.sh(開(kāi)辟資源)+flinkrun(提交任務(wù))】的運(yùn)行方式。yarn-session命令的用法如下表所示：知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法選擇

命令行參數(shù)含義必選-n,--container<arg>分配多少個(gè)yarn容器(=taskmanager的數(shù)量)可選-D<arg>動(dòng)態(tài)屬性

可選-d,--detached-d,--detached可選-jm,--jobManagerMemory<arg>JobManager的內(nèi)存[inMB]可選-nm,--name在YARN上為一個(gè)自定義的應(yīng)用設(shè)置一個(gè)名字

可選-q,--query顯示yarn中可用的資源(內(nèi)存,cpu核數(shù))可選-qu,--queue<arg>指定YARN隊(duì)列.可選-s,--slots<arg>每個(gè)TaskManager使用的slots數(shù)量

可選-tm,--taskManagerMemory<arg>每個(gè)TaskManager的內(nèi)存[inMB]可選-z,--zookeeperNamespace<arg>針對(duì)HA模式在zookeeper上創(chuàng)建NameSpace可選-id,--applicationId<yarnAppId>YARN集群上的任務(wù)id，附著到一個(gè)后臺(tái)運(yùn)行的yarnsession中知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法【運(yùn)行示例1】啟動(dòng)一個(gè)一直運(yùn)行的flink集群/bin/yarn-session.sh-n2-jm1024-tm1024[-d]把任務(wù)附著到一個(gè)已存在的flinkyarnsession./bin/yarn-session.sh-idapplication_1463870264508_0029執(zhí)行任務(wù)./bin/flinkrun./examples/batch/WordCount.jar-inputhdfs://hadoop100:9000/LICENSE-outputhdfs://hadoop100:9000/wordcount-result.txt如果需要停止任務(wù)可以在web界面停止或者命令行執(zhí)行cancel命令。知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法2.方式二第二種方式是每提交一個(gè)任務(wù)就在yarn上面啟動(dòng)一個(gè)flink小集群，通常推薦使用這種方法運(yùn)行，任務(wù)運(yùn)行完了資源會(huì)自動(dòng)釋放。主要使用的命令是【flinkrun-myarn-cluster(開(kāi)辟資源+提交任務(wù))】，flinkrun命令的常用用法如下表所示：選擇

命令行參數(shù)含義必選-n,--container<arg>分配多少個(gè)yarn容器(=taskmanager的數(shù)量)可選-D<arg>動(dòng)態(tài)屬性可選-d,--detached獨(dú)立運(yùn)行可選-jm,--jobManagerMemory<arg>JobManager的內(nèi)存[inMB]可選-nm,--name在YARN上為一個(gè)自定義的應(yīng)用設(shè)置一個(gè)名字可選-q,--query顯示yarn中可用的資源(內(nèi)存,cpu核數(shù))可選-qu,--queue<arg>指定YARN隊(duì)列.可選-s,--slots<arg>每個(gè)TaskManager使用的slots數(shù)量可選-tm,--taskManagerMemory<arg>每個(gè)TaskManager的內(nèi)存[inMB]可選-z,--zookeeperNamespace<arg>針對(duì)HA模式在zookeeper上創(chuàng)建NameSpace可選-id,--applicationId<yarnAppId>YARN集群上的任務(wù)id，附著到一個(gè)后臺(tái)運(yùn)行的yarnsession中知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法flinkrun[OPTIONS]<jar-file><arguments>命令中run的操作參數(shù)主要包括以下3個(gè)：命令參數(shù)含義-c,--class<classname>如果沒(méi)有在jar包中指定入口類(lèi)，則需要在這里通過(guò)這個(gè)參數(shù)指定-m,--jobmanager<host:port>指定需要連接的jobmanager(主節(jié)點(diǎn))地址，使用這個(gè)參數(shù)可以指定一個(gè)不同于配置文件中的jobmanager

-p,--parallelism<parallelism>指定程序的并行度?？梢愿采w配置文件中的默認(rèn)值。知識(shí)點(diǎn)1：FlinkonYARN的運(yùn)行方法【案例1】默認(rèn)查找當(dāng)前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】啟動(dòng)一個(gè)新的yarn-session：./bin/flinkrun-myarn-cluster-yn2./examples/batch/WordCount.jar-inputhdfs://hostname:port/hello.txt-outputhdfs://hostname:port/result1【小提示】yarnsession命令行的選項(xiàng)也可以使用./bin/flink工具獲得。它們都有一個(gè)y或者yarn的前綴。例如：./bin/flinkrun-myarn-cluster-yn2./examples/batch/WordCount.jar知識(shí)點(diǎn)2：故障調(diào)試與恢復(fù)1.Flinkonyarn的故障恢復(fù)Flink的yarn客戶端通過(guò)下面的配置參數(shù)來(lái)控制容器的故障恢復(fù)。這些參數(shù)可以通過(guò)conf/flink-conf.yaml或者在啟動(dòng)yarnsession的時(shí)候通過(guò)-D參數(shù)來(lái)指定。yarn.reallocate-failed：這個(gè)參數(shù)控制了flink是否應(yīng)該重新分配失敗的taskmanager容器。默認(rèn)是true。yarn.maximum-failed-containers：applicationMaster可以接受的容器最大失敗次數(shù)，達(dá)到這個(gè)參數(shù)，就會(huì)認(rèn)為yarnsession失敗。默認(rèn)這個(gè)次數(shù)和初始化請(qǐng)求的taskmanager數(shù)量相等(-n參數(shù)指定的)。yarn.application-attempts：applicationMaster重試的次數(shù)。如果這個(gè)值被設(shè)置為1(默認(rèn)就是1)，當(dāng)applicationmaster失敗的時(shí)候，yarnsession也會(huì)失敗。設(shè)置一個(gè)比較大的值的話，yarn會(huì)嘗試重啟applicationMaster。知識(shí)點(diǎn)2：故障調(diào)試與恢復(fù)2.調(diào)試失敗的yarnsession一個(gè)Flinkyarnsession部署失敗可能會(huì)有很多原因。一個(gè)錯(cuò)誤的hadoop配置(hdfs權(quán)限，yarn配置)，版本不兼容(使用cdh中的hadoop運(yùn)行flink)，或者其他的錯(cuò)誤。flinkyarnsession部署失敗是由于它自身的原因，用戶必須依賴于yarn的日志來(lái)進(jìn)行分析。

最有用的就是yarnlogaggregation，用戶必須在yarn-site.xml文件中設(shè)置yarn.log-aggregation-enable屬性為true，才能啟動(dòng)此功能。啟用yarnlogaggregation之后，用戶可以通過(guò)下面的命令來(lái)查看一個(gè)失敗的yarnsession的所有詳細(xì)日志：yarnlogs-applicationId<applicationID>。如果錯(cuò)誤發(fā)生在運(yùn)行時(shí)(例如某個(gè)taskmanager停止工作了一段時(shí)間)，flinkyarnclient也會(huì)打印一些錯(cuò)誤信息在控制臺(tái)上。除此之外，yarnresourcemanager的web界面(默認(rèn)端口是8088)，也可以用于在yarn程序運(yùn)行期間或運(yùn)行失敗時(shí)查看日志定位問(wèn)題。FlinkonYARN部署流程【工作流程】

部署yarn運(yùn)行模式的Flink基本工作流程包括：1.Flink集群部署規(guī)劃；2.修改yarn配置文件；3.以Flinkonyarn模式提交運(yùn)行詞頻統(tǒng)計(jì)樣例程序；4.通過(guò)yarn的WebUI監(jiān)控Flink任務(wù)執(zhí)行情況。FlinkonYARN部署流程1.集群規(guī)劃

Flinkonyarn模式需要依賴Hadoop集群，3個(gè)節(jié)點(diǎn)的Flink集群規(guī)劃如下表所示：主機(jī)名節(jié)點(diǎn)環(huán)境用途master01CentOS7、JDK1.8、Hadoop3.1.1、Flink客戶端主節(jié)點(diǎn)worker01CentOS7、JDK1.8、Hadop3.1.1從節(jié)點(diǎn)1worker02CentOS7、JDK1.8、Hadoop3.1.1從節(jié)點(diǎn)2FlinkonYARN部署流程2.修改yarn配置文件修改Hadoop集群的yarn-site.xml配置文件，每個(gè)文件中增加以下配置內(nèi)容：

<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模式提交運(yùn)行詞頻統(tǒng)計(jì)樣例程序在終端通過(guò)flinkrun命令提交執(zhí)行詞頻統(tǒng)計(jì)樣例程序，命令行中通過(guò)--input選項(xiàng)指定統(tǒng)計(jì)的源文件為/root/anaconda-ks.cfg，執(zhí)行結(jié)果通過(guò)--output指定/root/out1，執(zhí)行命令及過(guò)程如下所示：[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