一、常見部署模式分類

1. 按是否依賴外部資源調(diào)度

1.1 Standalone 模式

獨(dú)立模式 (Standalone) 是獨(dú)立運(yùn)行的，不依賴任何外部的資源管理平臺(tái)，只需要運(yùn)行所有 Flink 組件服務(wù)

1.2 Yarn 模式

Yarn 模式是指客戶端把 Flink 應(yīng)用提交給 Yarn 的 ResourceManager, Yarn 的 ResourceManager 會(huì)在 Yarn 的 NodeManager 上創(chuàng)建容器。在這些容器上，Flink 會(huì)部署 JobManager 和 TaskManager 的實(shí)例，從而啟動(dòng)集群。Flink 會(huì)根據(jù)運(yùn)行在 JobManger 上的作業(yè)所需要的 Slot 數(shù)量動(dòng)態(tài)分配 TaskManager 資源

2. 按集群的生命周期和執(zhí)行位置

2.1 會(huì)話模式

• 會(huì)話模式 (Session Mode) 是指先啟動(dòng)一個(gè)集群，保持一個(gè)會(huì)話并且確定所有的資源，然后向集群提交作業(yè)，所有提交的作業(yè)會(huì)競(jìng)爭(zhēng)集群中的資源，從而會(huì)出現(xiàn)資源不足作業(yè)執(zhí)行失敗的情況
• 會(huì)話模式比較適合于單個(gè)規(guī)模小、執(zhí)行時(shí)間短的大量作業(yè)

2.2 單作業(yè)模式

• 單作業(yè)模式 (Per-Job Mode) 是指為每一個(gè)提交的作業(yè)啟動(dòng)一個(gè)集群，由客戶端運(yùn)行應(yīng)用程序，然后啟動(dòng)集群，作業(yè)被提交給 JobManager，進(jìn)而分發(fā)給 TaskManager 執(zhí)行。作業(yè)作業(yè)完成后，集群就會(huì)關(guān)閉，所有資源也會(huì)釋放。
• 單作業(yè)模式在生產(chǎn)環(huán)境運(yùn)行更加穩(wěn)定，所以是實(shí)際應(yīng)用的首選模式
• 單作業(yè)模式一般需要借助一些資源管理框架來啟動(dòng)集群，比如 YARN、Kubernetes

2.3 應(yīng)用模式

• 應(yīng)用模式 (Application Mode) 是指為每一個(gè)提交的應(yīng)用單獨(dú)啟動(dòng)一個(gè) JobManager，也就是創(chuàng)建一個(gè)集群。這個(gè) JobManager 只為執(zhí)行這一個(gè)應(yīng)用而存在，執(zhí)行結(jié)束之后 JobManager 也就關(guān)閉了。這一模式下沒有客戶端的存在
• 應(yīng)用模式與單作業(yè)模式，都是提交作業(yè)之后才創(chuàng)建集群；單作業(yè)模式是通過客戶端來提交作業(yè)的，客戶端解析出的每一個(gè)作業(yè)對(duì)應(yīng)一個(gè)集群；而應(yīng)用模式下，是直接由 JobManager 執(zhí)行應(yīng)用程序的，并且即使應(yīng)用包含了多個(gè)作業(yè)，也只創(chuàng)建一個(gè)集群

二、常見部署模式組合

• Standalone + 會(huì)話模式
• Standalone + 應(yīng)用模式
• Yarn + 會(huì)話模式
• Yarn + 單作業(yè)模式
• Yarn + 應(yīng)用模式

三、獨(dú)立模式安裝

1. 單節(jié)點(diǎn)安裝

• flink 下載地址：https://flink.apache.org/downloads/

• 下載 flink 安裝包：flink-1.10.1-bin-scala_2.12.tgz

• 將安裝包上傳到虛擬機(jī)節(jié)點(diǎn)并解壓縮

  tar -zxvf flink-1.10.1-bin-scala_2.12.tgz -C /opt/module
  cd /opt/module
  mv flink-1.10.1 flink
  

• 進(jìn)入 flink 安裝目錄，執(zhí)行啟動(dòng)命令，并查看進(jìn)程

  cd /opt/module/flink
  bin/start-cluster.shjps
  

• 訪問 http://hadoop102:8081 進(jìn)入 flink 集群和任務(wù)監(jiān)控管理 Web 頁面

• 關(guān)閉 flink：bin/stop-cluster.sh

2. 集群安裝

2.1 集群規(guī)劃

節(jié)點(diǎn)服務(wù)器	hadoop102	hadoop103	hadoop104
角色	JobManager	TaskManager	TaskManager

2.2 步驟

• 下載 flink 安裝包：flink-1.10.1-bin-scala_2.12.tgz

• 將安裝包上傳到 hadoop102 并解壓縮

  tar -zxvf flink-1.10.1-bin-scala_2.12.tgz -C /opt/module
  cd /opt/module
  mv flink-1.10.1 flink
  

• 進(jìn)入 flink 安裝目錄下的 conf 目錄，修改配置文件 flink-conf.yaml

  cd /opt/module/flink/conf
  vim flink-conf.yaml
  

  #修改 jobmanager 內(nèi)部通信主機(jī)名
  jobmanager.rpc.address: hadoop102
  jobmanager.rpc.port: 6123
  

• 修改 conf 目錄下的 slaves 文件，配置 taskmanager 節(jié)點(diǎn)

  vim slaves #1.13版本為 workers 文件
  #添加內(nèi)容
  hadoop103
  hadoop104
  

• flink-conf.yaml 文件中常用配置項(xiàng)：

  #對(duì) JobManager 進(jìn)程可使用到的全部?jī)?nèi)存進(jìn)行配置，包括 JVM 元空間和其他開銷，默認(rèn)為 1600M，可以根據(jù)集群規(guī)模進(jìn)行適當(dāng)調(diào)整
  jobmanager.memory.process.size: 1600#對(duì) TaskManager 進(jìn)程可使用到的全部?jī)?nèi)存進(jìn)行配置，包括 JVM 元空間和其他開銷，默認(rèn)為 1600M，可以根據(jù)集群規(guī)模進(jìn)行適當(dāng)調(diào)整
  taskmanager.memory.process.size: 1600#對(duì)每個(gè) TaskManager 能夠分配的 TaskSlot 數(shù)量進(jìn)行配置，默認(rèn)為 1，可根據(jù) TaskManager 所在的機(jī)器能夠提供給 Flink 的 CPU 數(shù)量決定。Slot 就是 TaskManager 中具體運(yùn)行一個(gè)任務(wù)所分配的計(jì)算資源
  taskmanager.numberOfTaskSlots: 1#Flink 任務(wù)執(zhí)行的默認(rèn)并行度，優(yōu)先級(jí)低于代碼中進(jìn)行的并行度配置和任務(wù)提交時(shí)使用參數(shù)指定的并行度數(shù)量
  parallelism.default: 1
  

• 分發(fā) flink 安裝目錄到 hadoop103 和 hadoop104

• 在 hadoop102 上啟動(dòng) flink 集群

  cd /opt/module/flink
  bin/start-cluster.shjps
  

• 訪問 http://hadoop102:8081 查看 flink 監(jiān)控頁面

3. 提交任務(wù)

3.1 會(huì)話模式提交

• 啟動(dòng) Flink 集群服務(wù)

• 在 hadoop102 中啟動(dòng) netcat 網(wǎng)絡(luò)端口

  nc -lk 7777
  

• Web 頁面提交

  • 將編碼好的 Flink maven 工程打成 jar 包
  • 訪問 http://hadoop102:8081 進(jìn)入 flink 監(jiān)控頁面，選擇左側(cè)的 Submit New Job 選項(xiàng)菜單
  • 點(diǎn)擊 +Add New 按鈕，然后選擇 jar 包進(jìn)行上傳
  • 點(diǎn)擊頁面上上傳好的 jar 包項(xiàng)，配置填寫主程序類全類名、啟動(dòng)參數(shù)項(xiàng)、并行度等；點(diǎn)擊 submit 提交任務(wù)
  • 在頁面左側(cè)的 overview 和 jobs 等菜單選項(xiàng)下查看任務(wù)運(yùn)行情況
  • 一個(gè) job 所占據(jù)的 TaskSlots 數(shù)等于該 job 中最大的并行度

• 命令行提交

  #提交任務(wù)：bin/flink run -m [jobmanager主機(jī)和端口] -c [主程序類全類名] -p [并行度] [jar包的絕對(duì)路徑] [--param1 value1 --param2 value2 ...]
  cd flink
  bin/flink run -m hadoop102:8081 -c com.app.wc.StreamWordCount2 -p 3 /project/FlinkTutorial/target/FlinkTutorial-1.0-SNAPSHOT.jar --host localhost --port 7777#查看job：-a 可以查看已經(jīng)取消的job
  bin/flink list [-a]#取消job
  bin/flink cancel [jobId]
  

3.2 應(yīng)用模式提交

• 不能使用 start-cluster.sh 命令啟動(dòng)集群

• 將編碼好的 Flink maven 工程打成 jar 包，并將 jar 包上傳到 flink 安裝目錄下的 lib 目錄

• 啟動(dòng) JobManager

  cd /opt/module/flink
  bin/standalone-job.sh start --job-classname com.app.wc.StreamWordCount2
  

• 啟動(dòng) TaskManager

  cd /opt/module/flink
  bin/taskmanager.sh start
  

• 訪問 http://hadoop102:8081 查看 flink 監(jiān)控頁面的作業(yè)執(zhí)行

• 關(guān)閉

  cd /opt/module/flink
  bin/standalone-job.sh stop
  bin/taskmanager.sh stop
  

4. 高可用集群安裝

4.1 集群規(guī)劃

節(jié)點(diǎn)服務(wù)器	hadoop102	hadoop103	hadoop104
角色	JobManager	JobManager / TaskManager	TaskManager

4.2 步驟

• 在 hadoop102 上進(jìn)入 flink 目錄下的 conf 目錄，修改 flink-conf.yaml 文件

  cd /opt/module/flink/conf
  vim flink-conf.yaml
  

  #添加內(nèi)容
  high-availability: zookeeper
  high-availability.storageDir: hdfs://hadoop102:9820/flink/standalone/ha
  high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
  high-availability.zookeeper.path.root: /flink-standalone
  high-availability.cluster-id: /cluster_atguigu
  

• 修改 masters 文件

  vim masters
  #添加
  hadoop102:8081
  hadoop103:8081
  

• 分發(fā)配置到其他節(jié)點(diǎn)

• 啟動(dòng)

  #保證hadoop環(huán)境變量配置生效
  #啟動(dòng) hadoop 集群和 Zookeeper 集群
  #啟動(dòng) flink 集群
  cd /opt/module/flink
  bin/start-cluster.sh
  

• 訪問：http://hadoop102:8081 和 http://hadoop103:8081

四、Yarn 模式安裝

1. 安裝步驟

• 下載安裝包：

  • flink 1.8 版本之前可以直接下載基于 hadoop 版本編譯的安裝包
  • flink 1.8 及之后的版本除了下載基本的安裝包之外，還需要下載 Hadoop 相關(guān)版本的組件，如 flink-shaded-hadoop-2-uber-2.7.5-10.0.jar，并將該組件上傳至 Flink 的 lib 目錄下
  • flink 1.11 版本之后，不再提供 flink-shaded-hadoop-* 的 jar 包，而是通過配置環(huán)境變量完成與 Yarn 集群的對(duì)接

• 將安裝包上傳到虛擬機(jī)并解壓縮

  tar -zxvf flink-1.10.1-bin-scala_2.12.tgz -C /opt/module
  cd /opt/module
  mv flink-1.10.1 flink
  

• 進(jìn)入 conf 目錄，修改 flink-conf.yaml 文件

  cd /opt/module/flink/conf
  vim flink-conf.yaml
  

  #修改
  jobmanager.memory.process.size: 1600m
  taskmanager.memory.process.size: 1728m
  taskmanager.numberOfTaskSlots: 8
  parallelism.default: 1
  

• 確保正確安裝 Hadoop 集群和配置 Hadoop 環(huán)境變量

  sudo vim /etc/profile.d/my_env.shexport HADOOP_HOME=/opt/module/hadoop
  export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
  export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
  export HADOOP_CLASSPATH=`hadoop classpath`
  

2. 提交任務(wù)

2.1 會(huì)話模式提交

• 啟動(dòng) Hadoop 集群

• 執(zhí)行腳本命令向 Yarn 集群申請(qǐng)資源，開啟一個(gè) Yarn 會(huì)話，啟動(dòng) Flink 集群

  cd /opt/module/flink
  bin/yarn-session.sh -nm test#-n 參數(shù)：指定 TaskManager 數(shù)量
  #-s 參數(shù)：指定 slot 數(shù)量
  #-d：分離模式，如果你不想讓 Flink YARN 客戶端一直前臺(tái)運(yùn)行，可以使用這個(gè)參數(shù)，即使關(guān)掉當(dāng)前對(duì)話窗口，YARN session 也可以后臺(tái)運(yùn)行。
  #-jm(--jobManagerMemory)：配置 JobManager 所需內(nèi)存，默認(rèn)單位 MB。
  #-nm(--name)：配置在 YARN UI 界面上顯示的任務(wù)名。
  #-qu(--queue)：指定 YARN 隊(duì)列名。
  #-tm(--taskManager)：配置每個(gè) TaskManager 所使用內(nèi)存
  

• 提交作業(yè)

  • Web UI 提交，同獨(dú)立模式

  • 命令行提交：

    #1.將打包好的任務(wù)運(yùn)行 JAR 包上傳至集群
    #2.執(zhí)行命令
    cd /opt/module/flink
    bin/flink run -c com.app.wc.StreamWordCount2 FlinkTutorial-1.0-SNAPSHOT.jar
    

• 訪問 yarn Web UI 界面或 flink Web UI 界面查看作業(yè)執(zhí)行情況

2.2 單作業(yè)模式提交

• 啟動(dòng) Hadoop 集群

• 直接向 Yarn 提交一個(gè)單獨(dú)的作業(yè)，從而啟動(dòng)一個(gè) Flink 集群

  cd /opt/module/flink
  #命令一：
  bin/flink run -d -t yarn-per-job -c com.app.wc.StreamWordCount2 FlinkTutorial-1.0-SNAPSHOT.jar#命令二：
  bin/flink run -m yarn-cluster -c com.app.wc.StreamWordCount2 FlinkTutorial-1.0-SNAPSHOT.jar
  

• 訪問 yarn Web UI 界面或 flink Web UI 界面查看作業(yè)執(zhí)行情況

• 取消作業(yè)：

  cd /opt/module/flink
  bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
  bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>
  

2.3 應(yīng)用模式提交

• 啟動(dòng) Hadoop 集群

• 執(zhí)行命令提交作業(yè)

  #上傳 jar 包到集群
  cd /opt/module/flink
  bin/flink run-application -t yarn-application -c com.app.wc.StreamWordCount2 FlinkTutorial-1.0-SNAPSHOT.jar#上傳 jar 包到 hdfs
  bin/flink run-application -t yarn-application 
  -Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist-dir"
  hdfs://myhdfs/jars/my-application.jar
  

• 查看或取消作業(yè)

  /opt/module/flink
  bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
  bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>
  

3. 高可用配置

YARN 的高可用是只啟動(dòng)一個(gè) Jobmanager, 當(dāng)這個(gè) Jobmanager 掛了之后, YARN 會(huì)再次啟動(dòng)一個(gè), 本質(zhì)是利用的 YARN 的重試次數(shù)來實(shí)現(xiàn)的高可用

• 在 yarn-site.xml 中配置

  <property><name>yarn.resourcemanager.am.max-attempts</name><value>4</value><description>The maximum number of application master execution attempts.</description>
  </property>
  

• 分發(fā)配置到其他節(jié)點(diǎn)

• 在 flink-conf.yaml 中配置

  yarn.application-attempts: 3  #要小于 yarn 的重試次數(shù)
  high-availability: zookeeper
  high-availability.storageDir: hdfs://hadoop102:9820/flink/yarn/ha
  high-availability.zookeeper.quorum: hadoop102:2181,hadoop103:2181,hadoop104:2181
  high-availability.zookeeper.path.root: /flink-yarn
  

• 啟動(dòng) yarn-session，殺死 JobManager, 查看復(fù)活情況