Flink支持多種安裝模式。
- local（本地）——單機模式，一般不使用
- standalone——獨立模式，Flink自帶集群，開發測試環境使用
- yarn——計算資源統一由Hadoop YARN管理，生產環境測試
Standalone模式
步驟
1. 解壓flink壓縮包到指定目錄
2. 配置flink
3. 配置slaves節點
4. 分發flink到各個節點
5. 啟動集群
6. 提交WordCount程序測試
7. 查看Flink WebUI
具體操作
1. 上傳flink壓縮包到指定目錄
2. 解壓縮flink到 /export/servers 目錄
tar -xvzf flink-1.6.0-bin-hadoop26-scala_2.11.tgz -C /export/servers
3. 使用vi修改 conf/flink-conf.yaml
# 配置Master的機器名（IP地址）
jobmanager.rpc.address: node-1
# 配置每個taskmanager生成的臨時文件夾
taskmanager.tmp.dirs: /export/servers/flink-1.6.0/tmp
4. 使用vi修改slaves文件
node-1
node-2
node-3
5. 使用vi修改 /etc/profile 系統環境變量配置文件，添加HADOOP_CONF_DIR目錄
export HADOOP_CONF_DIR=/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
6. 分發/etc/profile到其他兩個節點
scp -r /etc/profile node-2:/etc
scp -r /etc/profile node-3:/etc
7. 每個節點重新加載環境變量
source /etc/profile
8. 使用scp命令分發flink到其他節點
scp -r /export/servers/flink-1.6.0/ node-2:/export/servers/
scp -r /export/servers/flink-1.6.0/ node-3:/export/servers/
9. 啟動Flink集群
10. 啟動HDFS集群
11. 在HDFS中創建/test/input目錄
hadoop fs -mkdir -p /test/input
12. 上傳wordcount.txt文件到HDFS /test/input目錄
hadoop fs -put /root/wordcount.txt /test/input
13. 並運行測試任務
bin/flink run /export/servers/flink-1.6.0/examples/batch/WordCount.jar --input hdfs://node-1:9000/test/input/wordcount.txt --output hdfs://node-1:9000/test/output/result.txt
14. 瀏覽Flink Web UI界面
http://node-1:8081
Standalone集群架構
- client客戶端提交任務給JobManager
- JobManager負責Flink集群計算資源管理，並分發任務給TaskManager執行
- TaskManager定期向JobManager匯報狀態
高可用HA模式
從上述架構圖中，可發現JobManager存在 單點故障 ，一旦JobManager出現意外，整個集群無法工作。所以，為了確 保集群的高可用，需要搭建Flink的HA。（如果是部署在YARN上，部署YARN的HA），我們這里演示如何搭建 Standalone 模式HA。
HA架構圖
步驟
1. 在flink-conf.yaml中添加zookeeper配置
2. 將配置過的HA的 flink-conf.yaml 分發到另外兩個節點
3. 分別到另外兩個節點中修改flink-conf.yaml中的配置
4. 在 masters 配置文件中添加多個節點
5. 分發masters配置文件到另外兩個節點
6. 啟動 zookeeper 集群
7. 啟動 flink 集群
具體操作
1. 在flink-conf.yaml中添加zookeeper配置
#開啟HA，使用文件系統作為快照存儲
state.backend: filesystem
#啟用檢查點，可以將快照保存到HDFS
state.backend.fs.checkpointdir: hdfs://node-1:9000/flink-checkpoints
#使用zookeeper搭建高可用
high-availability: zookeeper
# 存儲JobManager的元數據到HDFS
high-availability.storageDir: hdfs://node-1:9000/flink/ha/
high-availability.zookeeper.quorum: node-1:2181,node-2:2181,node-3:2181
2. 將配置過的HA的 flink-conf.yaml 分發到另外兩個節點
scp -r /export/servers/flink-1.6.0/conf/flink-conf.yaml node-2:/export/servers/flink-1.6.0/conf/
scp -r /export/servers/flink-1.6.0/conf/flink-conf.yaml node-3:/export/servers/flink-1.6.0/conf/
3. 到節點2中修改flink-conf.yaml中的配置，將JobManager設置為自己節點的名稱
jobmanager.rpc.address: node-2
4. 在 masters 配置文件中添加多個節點
node-1:8081
node-2:8082
5. 分發masters配置文件到另外兩個節點
scp /export/servers/flink-1.6.0/conf/masters node-2:/export/servers/flink-1.6.0/conf/
scp /export/servers/flink-1.6.0/conf/masters node-3:/export/servers/flink-1.6.0/conf/
6. 啟動 zookeeper 集群
7. 啟動 HDFS 集群
8. 啟動 flink 集群
9. 分別查看兩個節點的Flink Web UI
10. kill掉一個節點，查看另外的一個節點的Web UI
注意事項
切記搭建HA，需要將第二個節點的 jobmanager.rpc.address 修改為node-2
YARN模式
在企業中，經常需要將Flink集群部署到YARN，因為可以使用YARN來管理所有計算資源。而且Spark程序也可以部署到 YARN上。
Flink運行在YARN上，可以使用yarn-session來快速提交作業到YARN集群。
yarn-session提供兩種模式
1. 會話模式
- 使用Flink中的yarn-session（yarn客戶端），會啟動兩個必要服務 JobManager 和 TaskManagers
- 客戶端通過yarn-session提交作業
- yarn-session會一直啟動，不停地接收客戶端提交的作用
- 有大量的小作業，適合使用這種方式
2. 分離模式
- 直接提交任務給YARN
- 大作業，適合使用這種方式
步驟
1. 修改Hadoop的yarn-site.xml，添加該配置表示內存超過分配值，是否將任務殺掉。默認為true。
運行Flink程序，很容易超過分配的內存。
<property> <name>yarn.nodemanager.vmem-check-enabled</name> <value>false</value> </property>
2. 啟動HDFS、YARN集群
3. 使用yarn-session的模式提交作業
YARN Session模式
1. 在flink目錄啟動yarn-session
bin/yarn-session.sh -n 2 -tm 800 -s 1 -d
# -n 表示申請2個容器，
# -s 表示每個容器啟動多少個slot
# -tm 表示每個TaskManager申請800M內存
# -d 表示以后台程序方式運行
2. 使用flink提交任務
bin/flink run examples/batch/WordCount.jar
3. 如果程序運行完了，可以使用 yarn application -kill application_id 殺掉任務
yarn application -kill application_1554377097889_0002
分離模式
1. 使用flink直接提交任務
bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar # -yn 表示TaskManager的個數
2. 查看WEB UI