Flink(二)CentOS7.5搭建Flink1.6.1分布式集群
一. Flink的下載
安裝包下載地址:http://flink.apache.org/downloads.html ,選擇對應Hadoop的Flink版本下載
[admin@node21 software]$ wget http://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.6.1/flink-1.6.1-bin-hadoop27-scala_2.11.tgz [admin@node21 software]$ ll -rw-rw-r-- 1 admin admin 301867081 Sep 15 15:47 flink-1.6.1-bin-hadoop27-scala_2.11.tgz
Flink 有三種部署模式,分別是 Local、Standalone Cluster 和 Yarn Cluster。
二. Local模式
對於 Local 模式來說,JobManager 和 TaskManager 會公用一個 JVM 來完成 Workload。如果要驗證一個簡單的應用,Local 模式是最方便的。實際應用中大多使用 Standalone 或者 Yarn Cluster,而local模式只是將安裝包解壓啟動(./bin/start-local.sh)即可,在這裏不在演示。
三. Standalone 模式
快速入門教程地址:https://ci.apache.org/projects/flink/flink-docs-release-1.6/quickstart/setup_quickstart.html
1. 軟件要求
- Java 1.8.x或更高版本,
- ssh(必須運行sshd才能使用管理遠程組件的Flink腳本)
集群部署規劃
節點名稱 | master | worker | zookeeper |
node21 | master | zookeeper | |
node22 | master | worker | zookeeper |
node23 | worker | zookeeper |
2. 解壓
[admin@node21 software]$ tar zxvf flink-1.6.1-bin-hadoop27-scala_2.11.tgz -C /opt/module/ [admin@node21 software]$ cd /opt/module/ [admin@node21 module]$ ll drwxr-xr-x 8 admin admin 125 Sep 15 04:47 flink-1.6.1
3. 修改配置文件
[admin@node21 conf]$ ls flink-conf.yaml log4j-console.properties log4j-yarn-session.properties logback.xml masters sql-client-defaults.yaml log4j-cli.properties log4j.properties logback-console.xml logback-yarn.xml slaves zoo.cfg
修改flink/conf/masters,slaves,flink-conf.yaml
[admin@node21 conf]$ sudo vi masters node21:8081 [admin@node21 conf]$ sudo vi slaves node22 node23 [admin@node21 conf]$ sudo vi flink-conf.yaml taskmanager.numberOfTaskSlots:2 jobmanager.rpc.address: node21
可選配置:
- 每個JobManager(
jobmanager.heap.mb
)的可用內存量, - 每個TaskManager(
taskmanager.heap.mb
)的可用內存量, - 每臺機器的可用CPU數量(
taskmanager.numberOfTaskSlots
), - 集群中的CPU總數(
parallelism.default
)和 - 臨時目錄(
taskmanager.tmp.dirs
)
4. 拷貝安裝包到各節點
[admin@node21 module]$ scp -r flink-1.6.1/ admin@node22:`pwd` [admin@node21 module]$ scp -r flink-1.6.1/ admin@node23:`pwd`
5. 配置環境變量
配置所有節點Flink的環境變量
[admin@node21 flink-1.6.1]$ sudo vi /etc/profile export FLINK_HOME=/opt/module/flink-1.6.1 export PATH=$PATH:$FLINK_HOME/bin [admin@node21 flink-1.6.1]$ source /etc/profile
6. 啟動flink
[admin@node21 flink-1.6.1]$ ./bin/start-cluster.sh Starting cluster. Starting standalonesession daemon on host node21. Starting taskexecutor daemon on host node22. Starting taskexecutor daemon on host node23.
jps查看進程
7. WebUI查看
http://node21:8081
8. Flink 的 HA
首先,我們需要知道 Flink 有兩種部署的模式,分別是 Standalone 以及 Yarn Cluster 模式。對於 Standalone 來說,Flink 必須依賴於 Zookeeper 來實現 JobManager 的 HA(Zookeeper 已經成為了大部分開源框架 HA 必不可少的模塊)。在 Zookeeper 的幫助下,一個 Standalone 的 Flink 集群會同時有多個活著的 JobManager,其中只有一個處於工作狀態,其他處於 Standby 狀態。當工作中的 JobManager 失去連接後(如宕機或 Crash),Zookeeper 會從 Standby 中選舉新的 JobManager 來接管 Flink 集群。
對於 Yarn Cluaster 模式來說,Flink 就要依靠 Yarn 本身來對 JobManager 做 HA 了。其實這裏完全是 Yarn 的機制。對於 Yarn Cluster 模式來說,JobManager 和 TaskManager 都是被 Yarn 啟動在 Yarn 的 Container 中。此時的 JobManager,其實應該稱之為 Flink Application Master。也就說它的故障恢復,就完全依靠著 Yarn 中的 ResourceManager(和 MapReduce 的 AppMaster 一樣)。由於完全依賴了 Yarn,因此不同版本的 Yarn 可能會有細微的差異。這裏不再做深究。
1) 修改配置文件
修改flink-conf.yaml
high-availability:zookeeper #指定高可用模式(必須) high-availability.zookeeper.quorum:node21:2181,node22:2181,node23:2181 #ZooKeeper仲裁是ZooKeeper服務器的復制組,它提供分布式協調服務(必須) high-availability.storageDir:hdfs:///flink/haRecovery/ #JobManager元數據保存在文件系統storageDir中,只有指向此狀態的指針存儲在ZooKeeper中(必須) high-availability.zookeeper.path.root:/flink #根ZooKeeper節點,在該節點下放置所有集群節點(推薦)
high-availability.cluster-id:/flinkCluster #自定義集群(推薦)
修改conf/zoo.cfg
server.1=node21:2888:3888 server.2=node22:2888:3888 server.3=node23:2888:3888
修改conf/masters
node21:8081 node22:8081
修改slaves
node22
node23
同步配置文件conf到各節點
2) 啟動HA
先啟動zookeeper集群各節點(測試環境中也可以用Flink自帶的start-zookeeper-quorum.sh),啟動dfs ,再啟動flink
[admin@node21 flink-1.6.1]$ start-cluster.sh
3) 驗證HA
手動殺死node21上的master,此時,node22上的備用master轉為主mater。
4)手動將JobManager / TaskManager實例添加到群集
您可以使用bin/jobmanager.sh
和bin/taskmanager.sh
腳本將JobManager和TaskManager實例添加到正在運行的集群中。
添加JobManager
bin/jobmanager.sh ((start|start-foreground) [host] [webui-port])|stop|stop-all
添加TaskManager
bin/taskmanager.sh start|start-foreground|stop|stop-all
[admin@node21 flink-1.6.1]$ jobmanager.sh start node21
新添加的為從master。
9. 運行測試任務
[admin@node21 flink-1.6.1]$ flink run -m node21:8081 ./examples/batch/WordCount.jar --input /opt/wcinput/wc.txt --output /opt/wcoutput/ [admin@node21 flink-1.6.1]$ flink run -m node21:8081 ./examples/batch/WordCount.jar --input hdfs:///user/admin/input/wc.txt --output hdfs:///user/admin/output2
四. Yarn Cluster模式
1. 引入
在一個企業中,為了最大化的利用集群資源,一般都會在一個集群中同時運行多種類型的 Workload。因此 Flink 也支持在 Yarn 上面運行。首先,讓我們通過下圖了解下 Yarn 和 Flink 的關系。
在圖中可以看出,Flink 與 Yarn 的關系與 MapReduce 和 Yarn 的關系是一樣的。Flink 通過 Yarn 的接口實現了自己的 App Master。當在 Yarn 中部署了 Flink,Yarn 就會用自己的 Container 來啟動 Flink 的 JobManager(也就是 App Master)和 TaskManager。
啟動新的Flink YARN會話時,客戶端首先檢查所請求的資源(容器和內存)是否可用。之後,它將包含Flink和配置的jar上傳到HDFS(步驟1)。
客戶端的下一步是請求(步驟2)YARN容器以啟動ApplicationMaster(步驟3)。由於客戶端將配置和jar文件註冊為容器的資源,因此在該特定機器上運行的YARN的NodeManager將負責準備容器(例如,下載文件)。完成後,將啟動ApplicationMaster(AM)。
該JobManager和AM在同一容器中運行。一旦它們成功啟動,AM就知道JobManager(它自己的主機)的地址。它正在為TaskManagers生成一個新的Flink配置文件(以便它們可以連接到JobManager)。該文件也上傳到HDFS。此外,AM容器還提供Flink的Web界面。YARN代碼分配的所有端口都是臨時端口。這允許用戶並行執行多個Flink YARN會話。
之後,AM開始為Flink的TaskManagers分配容器,這將從HDFS下載jar文件和修改後的配置。完成這些步驟後,即可建立Flink並準備接受作業。
2. 修改環境變量
export HADOOP_CONF_DIR= /opt/module/hadoop-2.7.6/etc/hadoop
3. 部署啟動
[admin@node21 flink-1.6.1]$ yarn-session.sh -d -s 2 -tm 800 -n 2-n : TaskManager的數量,相當於executor的數量 -s : 每個JobManager的core的數量,executor-cores。建議將slot的數量設置每臺機器的處理器數量 -tm : 每個TaskManager的內存大小,executor-memory -jm : JobManager的內存大小,driver-memory
上面的命令的意思是,同時向Yarn申請3個container,其中 2 個 Container 啟動 TaskManager(-n 2),每個 TaskManager 擁有兩個 Task Slot(-s 2),並且向每個 TaskManager 的 Container 申請 800M 的內存,以及一個ApplicationMaster(Job Manager)。
Flink部署到Yarn Cluster後,會顯示Job Manager的連接細節信息。
Flink on Yarn會覆蓋下面幾個參數,如果不希望改變配置文件中的參數,可以動態的通過-D選項指定,如 -Dfs.overwrite-files=true -Dtaskmanager.network.numberOfBuffers=16368
jobmanager.rpc.address:因為JobManager會經常分配到不同的機器上
taskmanager.tmp.dirs:使用Yarn提供的tmp目錄
parallelism.default:如果有指定slot個數的情況下
yarn-session.sh會掛起進程,所以可以通過在終端使用CTRL+C或輸入stop停止yarn-session。
如果不希望Flink Yarn client長期運行,Flink提供了一種detached YARN session,啟動時候加上參數-d或—detached
在上面的命令成功後,我們就可以在 Yarn Application 頁面看到 Flink 的紀錄。如下圖。
如果在虛擬機中測試,可能會遇到錯誤。這裏需要註意內存的大小,Flink 向 Yarn 會申請多個 Container,但是 Yarn 的配置可能限制了 Container 所能申請的內存大小,甚至 Yarn 本身所管理的內存就很小。這樣很可能無法正常啟動 TaskManager,尤其當指定多個 TaskManager 的時候。因此,在啟動 Flink 之後,需要去 Flink 的頁面中檢查下 Flink 的狀態。這裏可以從 RM 的頁面中,直接跳轉(點擊 Tracking UI)。這時候 Flink 的頁面如圖
yarn-session.sh啟動命令參數如下:
[admin@node21 flink-1.6.1]$ yarn-session.sh --help Usage: Required -n,--container <arg> Number of YARN container to allocate (=Number of Task Managers) Optional -D <property=value> 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 <arg> Attach to running YARN session -j,--jar <arg> Path to Flink jar file -jm,--jobManagerMemory <arg> Memory for JobManager Container with optional unit (default: 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 i n the configuration. -n,--container <arg> Number of YARN container to allocate (=Number of Task Managers) -nl,--nodeLabel <arg> Specify YARN node label for the YARN application -nm,--name <arg> Set a custom name for the application on YARN -q,--query Display available YARN resources (memory, cores) -qu,--queue <arg> Specify YARN queue. -s,--slots <arg> Number of slots per TaskManager -st,--streaming Start Flink in streaming mode -t,--ship <arg> Ship files in the specified directory (t for transfer) -tm,--taskManagerMemory <arg> 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 <arg> Namespace to create the Zookeeper sub-paths for high availability mode
4. 提交任務
之後,我們可以通過這種方式提交我們的任務
[admin@node21 flink-1.6.1]$ ./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar --input /opt/wcinput/wc.txt --output /opt/wcoutput/
以上命令在參數前加上y前綴,-yn表示TaskManager個數。
在這個模式下,同樣可以使用-m yarn-cluster提交一個"運行後即焚"的detached yarn(-yd)作業到yarn cluster。
5. 停止yarn cluster
yarn application -kill application_1539058959130_0001
五.錯誤異常
1.超時
Caused by: java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id 96ab53853de9c9e56035d140d849c3bd timed out.
Flink(二)CentOS7.5搭建Flink1.6.1分布式集群