Flink集群部署

部署方式

一般來講有三種方式：

Local
Standalone
Flink On Yarn/Mesos/K8s…

單機模式

參考上一篇Flink從入門到放棄(入門篇2)-本地環境搭建&構建第一個Flink應用

Standalone模式部署

我們基於CentOS7虛擬機搭建一個3個節點的集群：

角色分配：

Master: 192.168.246.134
Slave: 192.168.246.135
Slave: 192.168.246.136

復制代碼

192.168.246.134 jobmanager
192.168.246.135 taskmanager
192.168.246.136 taskmanager
復制代碼

假設三台機器都存在：用戶root 密碼為123

192.168.246.134 master
192.168.246.135 slave1
192.168.246.136 slave2

復制代碼

三台機器首先要做ssh免登，具體方法很簡單，可以百度。

下載一個包到本地：這里我選擇了1.7.2版本+Hadoop2.8+Scala2.11版本然后，分發

scp flink-1.7.2-bin-hadoop28-scala_2.11.tgz root@192.168.246.13X:~
scp jdk-8u11-linux-x64.tar.gz root@192.168.246.13X:~
注意：X代表4、5、6，分發到3台機器

修改解壓后目錄屬主：
Chown -R  root:root flink/
Chown -R root:root jdk8/

export JAVA_HOME=/root/jdk8 export JRE_HOME=${JAVA_HOME}/jre export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib export PATH=${JAVA_HOME}/bin:$PATH 復制代碼

分別修改master和slave的flink-conf.yaml文件

Vim  flink/conf/flink-conf.yaml

##配置master節點ip jobmanager.rpc.address: 192.168.1.100 ##配置slave節點可用內存，單位MB taskmanager.heap.mb: 25600 ##配置每個節點的可用slot，1 核CPU對應 1 slot ##the number of available CPUs per machine taskmanager.numberOfTaskSlots: 30 ##默認並行度 1 slot資源 parallelism.default: 1 修改slave節點配置文件slaves： 192.168.246.135 192.168.246.136 復制代碼

啟動集群：

##在master節點上執行此腳本，就可以啟動集群，前提要保證master節點到slaver節點可以免密登錄， ##因為它的啟動過程是：先在master節點啟動jobmanager進程，然后ssh到各slaver節點啟動taskmanager進程 ./bin/start-cluster.sh 停止集群： ./bin/stop-cluster.sh 復制代碼

Flink on yarn集群部署

Yarn的簡介：

ResourceManager ResourceManager 負責整個集群的資源管理和分配，是一個全局的資源管理系統。 NodeManager 以心跳的方式向 ResourceManager 匯報資源使用情況（目前主要是 CPU 和內存的使用情況）。RM 只接受 NM 的資源回報信息，對於具體的資源處理則交給 NM 自己處理。
NodeManager NodeManager 是每個節點上的資源和任務管理器，它是管理這台機器的代理，負責該節點程序的運行，以及該節點資源的管理和監控。YARN 集群每個節點都運行一個NodeManager。 NodeManager 定時向 ResourceManager 匯報本節點資源（CPU、內存）的使用情況和Container 的運行狀態。當 ResourceManager 宕機時 NodeManager 自動連接 RM 備用節點。 NodeManager 接收並處理來自 ApplicationMaster 的 Container 啟動、停止等各種請求。
ApplicationMaster 負責與 RM 調度器協商以獲取資源（用 Container 表示）。將得到的任務進一步分配給內部的任務(資源的二次分配)。與 NM 通信以啟動/停止任務。監控所有任務運行狀態，並在任務運行失敗時重新為任務申請資源以重啟任務

Flink on yarn 集群啟動步驟

步驟1 用戶向YARN中提交應用程序，其中包括ApplicationMaster程序、啟動ApplicationMaster的命令、用戶程序等。
步驟2 ResourceManager為該應用程序分配第一個Container，並與對應的Node-Manager通信，要求它在這個Container中啟動應用程序的ApplicationMaster。
步驟3 ApplicationMaster首先向ResourceManager注冊，這樣用戶可以直接通過ResourceManager查看應用程序的運行狀態，然后它將為各個任務申請資源，並監控它的運行狀態，直到運行結束，即重復步驟4~7。
步驟4 ApplicationMaster采用輪詢的方式通過RPC協議向ResourceManager申請和領取資源。
步驟5 一旦ApplicationMaster申請到資源后，便與對應的NodeManager通信，要求它啟動任務。
步驟6 NodeManager為任務設置好運行環境（包括環境變量、JAR包、二進制程序等）后，將任務啟動命令寫到一個腳本中，並通過運行該腳本啟動任務。
步驟7 各個任務通過某個RPC協議向ApplicationMaster匯報自己的狀態和進度，以讓ApplicationMaster隨時掌握各個任務的運行狀態，從而可以在任務失敗時重新啟動任務。在應用程序運行過程中，用戶可隨時通過RPC向ApplicationMaster查詢應用程序的當前運行狀態。
步驟8 應用程序運行完成后，ApplicationMaster向ResourceManager注銷並關閉自己

on yarn 集群部署

設置Hadoop環境變量：

[root@hadoop2 flink-1.7.2]# vi /etc/profile export HADOOP_CONF_DIR=這里是你自己的hadoop路徑 復制代碼

bin/yarn-session.sh -h 查看使用方法:

bin/yarn-session.sh -h

Usage:
Required
-n,--container <arg> 為YARN分配容器的數量 (=Number of Task Managers)
Optional
-D <property=value> 動態屬性 
-d,--detached 以分離模式運行作業 -h,--help Yarn session幫助. -id,--applicationId <arg> 連接到一個正在運行的YARN session -j,--jar <arg> Flink jar文件的路徑 -jm,--jobManagerMemory <arg> JobManager的內存大小，driver-memory [in MB] -m,--jobmanager <arg> Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration. -n,--container <arg> TaskManager的數量，相當於executor的數量 -nm,--name <arg> 設置YARN應用自定義名稱 -q,--query 顯示可用的YARN資源 (memory, cores) -qu,--queue <arg> 指定YARN隊列 -s,--slots <arg> 每個JobManager的core的數量，executor-cores。建議將slot的數量設置每台機器的處理器數量 -st,--streaming 在流模式下啟動Flink -t,--ship <arg> 在指定目錄中傳送文件(t for transfer) -tm,--taskManagerMemory <arg> 每個TaskManager的內存大小，executor-memory [in MB] -yd,--yarndetached 如果存在，則以分離模式運行作業 (deprecated; use non-YARN specific option instead) -z,--zookeeperNamespace <arg> 為高可用性模式創建Zookeeper子路徑的命名空間 復制代碼

在啟動的是可以指定TaskManager的個數以及內存(默認是1G)，也可以指定JobManager的內存，但是JobManager的個數只能是一個

我們開啟動一個YARN session：

./bin/yarn-session.sh -n 4 -tm 8192 -s 8 復制代碼

上面命令啟動了4個TaskManager，每個TaskManager內存為8G且占用了8個核(是每個TaskManager，默認是1個核)。在啟動YARN session的時候會加載conf/flink-config.yaml配置文件，我們可以根據自己的需求去修改里面的相關參數.

YARN session啟動之后就可以使用bin/flink來啟動提交作業:

例如：

./bin/flink run -c com.demo.wangzhiwu.WordCount $DEMO_DIR/target/flink-demo-1.0.SNAPSHOT.jar --port 9000 復制代碼

flink run的用法如下：

用法: run [OPTIONS] <jar-file> <arguments>  
  "run" 操作參數: -c,--class <classname> 如果沒有在jar包中指定入口類，則需要在這里通過這個參數指定 -m,--jobmanager <host:port> 指定需要連接的jobmanager(主節點)地址 使用這個參數可以指定一個不同於配置文件中的jobmanager -p,--parallelism <parallelism> 指定程序的並行度。可以覆蓋配置文件中的默認值。 復制代碼

使用run 命令向yarn集群提交一個job。客戶端可以確定jobmanager的地址。當然，你也可以通過-m參數指定jobmanager。jobmanager的地址在yarn控制台上可以看到。

值得注意的是：

上面的YARN session是在Hadoop YARN環境下啟動一個Flink cluster集群，里面的資源是可以共享給其他的Flink作業。我們還可以在YARN上啟動一個Flink作業。這里我們還是使用./bin/flink，但是不需要事先啟動YARN session：

./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar      \
          --input hdfs://user/hadoop/input.txt                            \
          --output hdfs://user/hadoop/output.txt
復制代碼

上面的命令同樣會啟動一個類似於YARN session啟動的頁面。其中的-yn是指TaskManager的個數，必須要指定。

后台運行 yarn session

如果你不希望flink yarn client一直運行，也可以啟動一個后台運行的yarn session。使用這個參數：-d 或者 --detached 在這種情況下，flink yarn client將會只提交任務到集群然后關閉自己。注意：在這種情況下，無法使用flink停止yarn session。必須使用yarn工具來停止yarn session

yarn application -kill <applicationId> 
復制代碼

flink on yarn的故障恢復

flink 的 yarn 客戶端通過下面的配置參數來控制容器的故障恢復。這些參數可以通過conf/flink-conf.yaml 或者在啟動yarn session的時候通過-D參數來指定。

yarn.reallocate-failed：這個參數控制了flink是否應該重新分配失敗的taskmanager容器。默認是true。
yarn.maximum-failed-containers：applicationMaster可以接受的容器最大失敗次數，達到這個參數，就會認為yarn session失敗。默認這個次數和初始化請求的taskmanager數量相等(-n 參數指定的)。
yarn.application-attempts：applicationMaster重試的次數。如果這個值被設置為1(默認就是1)，當application master失敗的時候，yarn session也會失敗。設置一個比較大的值的話，yarn會嘗試重啟applicationMaster。

日志文件查看

在某種情況下，flink yarn session 部署失敗是由於它自身的原因，用戶必須依賴於yarn的日志來進行分析。最有用的就是yarn log aggregation 。啟動它，用戶必須在yarn-site.xml文件中設置yarn.log-aggregation-enable 屬性為true。一旦啟用了，用戶可以通過下面的命令來查看一個失敗的yarn session的所有詳細日志。

yarn logs -applicationId <application ID>  
復制代碼

完。

公眾號推薦

全網唯一一個從0開始幫助Java開發者轉做大數據領域的公眾號~
大數據技術與架構或者搜索import_bigdata關注~
海量【java和大數據的面試題+視頻資料】整理在公眾號，關注后可以下載~