一、Flink環(huán)境搭建

1.1 flink部署方式

Flink可以選擇的部署方式有：
Local、Standalone（資源利用率低）、Yarn、Mesos、Docker、Kubernetes、AWS。
我們主要對Standalone模式和Yarn模式下的Flink集群部署進行分析。
Standalone模式常用于單機進行程序測試，Yarn模式常用于實際線上生產(chǎn)環(huán)境。

創(chuàng)新互聯(lián)建站專注于桂林網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗。 熱誠為您提供桂林營銷型網(wǎng)站建設(shè)，桂林網(wǎng)站制作、桂林網(wǎng)頁設(shè)計、桂林網(wǎng)站官網(wǎng)定制、微信平臺小程序開發(fā)服務(wù)，打造桂林網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供桂林網(wǎng)站排名全網(wǎng)營銷落地服務(wù)。

1.2 集群規(guī)劃

1、集群規(guī)劃

節(jié)點名稱	master(jobManager)	worker(taskManager)	zookeeper
bigdata11	√		√
bigdata21	√	√	√
bigdata31		√	√

（注：zookeeper只是用于實現(xiàn)master HA的必要組件，如果不需要master HA，則zookeeper可以去掉。）

2、軟件版本

jdk	1.8
scala	2.11.8
hadoop	2.8
zookeeper	3.4.10
flink	1.6.1

3、基礎(chǔ)環(huán)境
安裝好jdk、scala、hadoop（hdfs+yarn都要部署好）、zookeeper，部署方法看之前的相關(guān)文章。而且要注意的是，節(jié)點之間要配置好ssh秘鑰免登陸。

1.3 Standalone模式安裝

1、解壓程序：

tar -zxvf flink-1.6.1-bin-hadoop28-scala_2.11.tgz -C /opt/module/修改配置文件

2、修改配置文件

配置master節(jié)點地址：
[root@bigdata11 conf]$ sudo vi masters
bigdata11:8081

配置worker節(jié)點地址：
[root@bigdata11 conf]$ sudo vi slaves
bigdata12
bigdata13

修改flink工作參數(shù)：
[root@bigdata11 conf]$ sudo vi flink-conf.yaml 
taskmanager.numberOfTaskSlots：2   //52行
jobmanager.rpc.address: bigdata11  //33行  指定jobmanager 的rpc地址
可選配置：
??每個JobManager（jobmanager.heap.mb）的可用內(nèi)存量，
??每個TaskManager（taskmanager.heap.mb）的可用內(nèi)存量，
??每臺機器（taskManager）的可用的slot數(shù)量（taskmanager.numberOfTaskSlots），
??每個job的并行度（parallelism.default）
??臨時目錄（taskmanager.tmp.dirs）

3、配置環(huán)境變量

vim /etc/profile.d/flink.sh
export FLINK_HOME=/opt/module/flink-1.6.1
export PATH=$PATH:$FLINK_HOME/bin

然后source /etc/profile.d/flink.sh 啟用環(huán)境變量

4、拷貝配置好的/opt/module/flink-1.6.1到其他節(jié)點
使用scp或者rsync

scp -r /opt/module/flink-1.6.1 bigdata12:`pwd`
scp -r /opt/module/flink-1.6.1 bigdata13:`pwd`

同時配置好其他兩臺的環(huán)境變量

5、啟動flink集群

[root@bigdata11 flink-1.6.1]$ ./bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host bigdata11.
Starting taskexecutor daemon on host bigdata12.
Starting taskexecutor daemon on host bigdata13.

使用jps可以在對應(yīng)的節(jié)點上查看對應(yīng)的進程

StandloneSessionClusterEntrypoint  這是jobmanager進程
TaskManagerRunner   這是taskmanager進程

6、web UI 查看
http://bigdata11:8081

7、運行測試任務(wù)

flink run -m bigdata11:8081 ./examples/batch/WordCount.jar --input /opt/module/datas/word.txt --output /tmp/word.output

8、增減節(jié)點到集群中

增加/減少jobmanager節(jié)點：
bin/jobmanager.sh ((start|start-foreground) [host] [webui-port])|stop|stop-all

增加/減少taskmanager節(jié)點（需要到當(dāng)前節(jié)點去啟動）：
bin/taskmanager.sh start|start-foreground|stop|stop-all

1.4 standalone模式j(luò)obManager HA

? 首先，我們需要知道 Flink 有兩種部署的模式，分別是 Standalone 以及 Yarn Cluster 模式。對于 Standalone 來說，F(xiàn)link 必須依賴于 Zookeeper 來實現(xiàn) JobManager 的 HA（Zookeeper 已經(jīng)成為了大部分開源框架 HA 必不可少的模塊）。在 Zookeeper 的幫助下，一個 Standalone 的 Flink 集群會同時有多個活著的 JobManager，其中只有一個處于工作狀態(tài)，其他處于 Standby 狀態(tài)。當(dāng)工作中的 JobManager 失去連接后（如宕機或 Crash），Zookeeper 會從 Standby 中選舉新的 JobManager 來接管 Flink 集群。
? 對于 Yarn Cluaster 模式來說，F(xiàn)link 就要依靠 Yarn 本身來對 JobManager 做 HA 了。其實這里完全是 Yarn 的機制。對于 Yarn Cluster 模式來說，JobManager 和 TaskManager 都是被 Yarn 啟動在 Yarn 的 Container 中。此時的 JobManager，其實應(yīng)該稱之為 Flink Application Master。也就說它的故障恢復(fù)，就完全依靠著 Yarn 中的 ResourceManager（和 MapReduce 的 AppMaster 一樣）。由于完全依賴了 Yarn，因此不同版本的 Yarn 可能會有細微的差異。這里不再做深究。

1、修改配置文件
conf/flink-conf.yaml

注釋掉 
#jobmanager.rpc.address: bigdata11

修改下面的配置
high-availability: zookeeper   //73行 指定高可用方式為zookeeper

#指定高可用模式中zookeeper的地址列表 //88行
high-availability.zookeeper.quorum:bigdata11:2181,bigdata12:2181,bigdata13:2181

#指定將jobmanager狀態(tài)數(shù)據(jù)持久化保存到hdfs中
high-availability.storageDir: hdfs:///flink/ha/       

#JobManager元數(shù)據(jù)保存在文件系統(tǒng)storageDir中，只有指向此狀態(tài)的指針存儲在ZooKeeper中（必須） //沒有
high-availability.zookeeper.path.root: /flink         

#根ZooKeeper節(jié)點，在該節(jié)點下放置所有集群節(jié)點（推薦），這是集群節(jié)點信息保存位置
high-availability.cluster-id:/flinkCluster           

＃自定義集群（推薦），這里是檢查點和保存點的配置，保存在hdfs中，非必須
state.backend: filesystem
state.checkpoints.dir: hdfs:///flink/checkpoints
state.savepoints.dir: hdfs:///flink/checkpoints

conf/masters

將主備jobmanager地址都寫到該配置文件中。
bigdata11:8081
bigdata12:8081

conf/zoo.cfg

server.1=bigdata11:2888:3888
server.2=bigdata12:2888:3888
server.3=bigdata13:2888:3888

修改完后同步配置到其他所有節(jié)點中。

2、啟動集群

先啟動好zookeeper服務(wù)。
然后啟動hdfs服務(wù)。
最后啟動flink集群。 start-cluster.sh

1.5 yarn模式安裝
部署步驟和上面standalone基本一樣，這里不重復(fù)。還要添加以下配置：
配置好hadoop（hdfs和yarn）環(huán)境，同時配置好HADOOP_HOME這個環(huán)境變量。
接著在yarn下啟動jobmanager和taskmanager。

/opt/module/flink-1.6.1/bin/yarn-session.sh -n 2 -s 4 -jm 1024 -tm 1024 -nm test -d

其中：
-n(--container)：TaskManager的數(shù)量。
-s(--slots)：    每個TaskManager的slot數(shù)量，默認(rèn)一個slot一個core，默認(rèn)每個taskmanager的slot的個數(shù)為1，有時可以多一些taskmanager，做冗余。
-jm：JobManager的內(nèi)存（單位MB)。
-tm：每個taskmanager的內(nèi)存（單位MB)。
-nm：yarn 的appName(現(xiàn)在yarn的ui上的名字)。 
-d：后臺執(zhí)行

會自動根據(jù) conf/ 下的配置文件啟動對應(yīng)的jobmanager和taskmanager的

啟動完成后，可以到y(tǒng)arn 的web頁面查看到剛才提交會話任務(wù)：

http://bigdata11:8088

同時可以在提交session的節(jié)點上使用jps查看對應(yīng)的進程：

YarnSessionClusterEntrypoint  這個就是剛剛提交的yarn-session維持的session進程

提交測試任務(wù)到y(tǒng)arn中的flink集群運行

./bin/flink run ./examples/batch/WordCount.jar --input 輸出數(shù)據(jù)路徑
--output 輸出數(shù)據(jù)路徑

可以手動使用 -m jobManagerAddress 指定jobmanager地址，但是flink client可以自動根據(jù)flink的配置文件獲取到j(luò)obmanager地址，所以可以不用指定

提交任務(wù)之后，可以在yarn的web頁面中查看到相關(guān)的任務(wù)信息

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

本文名稱：二、flink--集群環(huán)境搭建-創(chuàng)新互聯(lián)
文章分享：http://www.rwnh.cn/article10/dscdgo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供響應(yīng)式網(wǎng)站、微信小程序、自適應(yīng)網(wǎng)站、建站公司、動態(tài)網(wǎng)站、App開發(fā)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)