Flink On Standalone 即Flink任務(wù)運(yùn)行在Standalone集群中，Standlone集群部署時(shí)采用Session模式來(lái)構(gòu)建集群，即：首先構(gòu)建一個(gè)Flink集群，F(xiàn)link集群資源就固定了，所有提交到該集群的Flink作業(yè)都運(yùn)行在這一個(gè)集群中，如果集群中提交的任務(wù)多資源不夠時(shí)，需要手動(dòng)增加節(jié)點(diǎn)，所以Flink 基于Standalone運(yùn)行任務(wù)一般用在開(kāi)發(fā)測(cè)試或者企業(yè)實(shí)時(shí)業(yè)務(wù)較少的場(chǎng)景下。
Flink On Standalone 任務(wù)提交支持Session會(huì)話模式和Application應(yīng)用模式，不支持Per-Job單作業(yè)模式。下面介紹基于Standalone 的Session會(huì)話模式和Application應(yīng)用模式任務(wù)提交命令和原理，演示兩類任務(wù)提交模式的代碼還是以上一章節(jié)中讀取Socket 數(shù)據(jù)進(jìn)行實(shí)時(shí)WordCount統(tǒng)計(jì)代碼為例，代碼如下：
package com.lanson.flinkjava.code.chapter4;import org.apache.flink.api.common.typeinfo.Types;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;/** * 讀取Socket數(shù)據(jù)進(jìn)行實(shí)時(shí)WordCount統(tǒng)計(jì) */public class SocketWordCount { public static void main(String[] args) throws Exception { //1.準(zhǔn)備環(huán)境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //2.讀取Socket數(shù)據(jù) DataStreamSource ds = env.socketTextStream("node3", 9999); //3.準(zhǔn)備K,V格式數(shù)據(jù) SingleOutputStreamOperator> tupleDS = ds.flatMap((String line, Collector> out) -> { String[] words = line.split(","); for (String word : words) { out.collect(Tuple2.of(word, 1)); } }).returns(Types.TUPLE(Types.STRING, Types.INT)); //4.聚合打印結(jié)果 tupleDS.keyBy(tp -> tp.f0).sum(1).print(); //5.execute觸發(fā)執(zhí)行 env.execute(); }} 將以上代碼進(jìn)行打包，名稱為"FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar"，并在node3節(jié)點(diǎn)上啟動(dòng)socket服務(wù)（nc -lk 9999）。
(資料圖)
在Standalone集群搭建完成后，基于Standalone集群提交Flink任務(wù)方式就是使用的Session模式，提交任務(wù)之前首先啟動(dòng)Standalone集群($FLINK_HOME/bin/start-cluster.sh)，然后再提交任務(wù)，Standalone Session模式提交任務(wù)命令如下：
[root@node1 ~]# cd /software/flink-1.16.0/bin/[root@node1 bin]# ./flink run -m node1:8081 -d -c com.lanson.flinkjava.code.chapter4.SocketWordCount /root/flink-jar-test/FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar以上提交任務(wù)的參數(shù)解釋如下：
參數(shù) | 解釋 |
|---|---|
-m | --jobmanager,指定提交任務(wù)連接的JobManager地址。 |
-c | --class,指定運(yùn)行的class主類。 |
-d | --detached，任務(wù)提交后在后臺(tái)獨(dú)立運(yùn)行，退出客戶端，也可不指定。 |
-p | --parallelism,執(zhí)行程序的并行度。 |
以上任務(wù)提交完成后，我們可以登錄Flink WebUI（https://node1:8081）查看啟動(dòng)一個(gè)任務(wù)：
再次按照以上命令提交Flink任務(wù)可以看到集群中會(huì)有2個(gè)任務(wù)，說(shuō)明Standalone Session模式下提交的所有Flink任務(wù)共享集群資源，如下：
以上提交Flink流任務(wù)的名稱默認(rèn)為"Flink Streaming Job"，也可以通過(guò)參數(shù)"pipeline.name"來(lái)自定義指定Job 名稱，提交命令如下：
./flink run -m node1:8081 -d -Dpipeline.name=socket-wc1 -c com.lanson.flinkjava.code.chapter4.SocketWordCount /root/flink-jar-test/FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar提交之后，可以看到頁(yè)面中有三個(gè)任務(wù)，最后一個(gè)任務(wù)提交的名稱改成了自定義任務(wù)名稱。
Standalone Session模式提交任務(wù)中首先需要?jiǎng)?chuàng)建Flink集群，集群創(chuàng)建啟動(dòng)的同時(shí)Dispatcher、JobMaster、ResourceManager對(duì)象一并創(chuàng)建、TaskManager也一并啟動(dòng)，TaskManager會(huì)向集群ResourceManager匯報(bào)Slot信息，F(xiàn)link集群資源也就確定了。Standalone Session模式提交任務(wù)流程如下：
在客戶端提交Flink任務(wù)，客戶端會(huì)將任務(wù)轉(zhuǎn)換成JobGraph提交給JobManager。Dispatcher將提交任務(wù)提交給JobMaster。JobMaster向ResourceManager申請(qǐng)Slot資源。ResourceManager會(huì)在對(duì)應(yīng)的TaskManager上劃分Slot資源。TaskManager向JobMaster offer Slot資源。JobMaster將任務(wù)對(duì)應(yīng)的task發(fā)送到TaskManager上執(zhí)行。Standalone Application模式中不會(huì)預(yù)先創(chuàng)建Flink集群，在提交Flink 任務(wù)的同時(shí)會(huì)創(chuàng)建JobManager，啟動(dòng)Flink集群,然后需要手動(dòng)啟動(dòng)TaskManager連接該Flink集群，啟動(dòng)的TaskManager會(huì)根據(jù)$FLINK_HOME/conf/flink-conf.yaml配置文件中的"jobmanager.rpc.address"配置找JobManager，所以這里選擇在node1節(jié)點(diǎn)上提交任務(wù)并啟動(dòng)JobManager，方便后續(xù)其他節(jié)點(diǎn)啟動(dòng)TaskManager后連接該節(jié)點(diǎn)。Standalone Appliction模式提交任務(wù)步驟和命令如下：
1.1、準(zhǔn)備Flink jar包
在node1節(jié)點(diǎn)上將Flink 打好的"FlinkJavaCode-1.0-SNAPSHOT-jar-with-dependencies.jar"jar包放在 $FLINK_HOME/lib目錄下。
1.2、提交任務(wù)，在node1 節(jié)點(diǎn)上啟動(dòng) JobManager
cd /software/flink-1.16.0/bin/#執(zhí)行如下命令，啟動(dòng)JobManager ./standalone-job.sh start --job-classname com.lanson.flinkjava.code.chapter4.SocketWordCount執(zhí)行以上命令后會(huì)自動(dòng)從$FLINK_HOME/lib中掃描所有jar包，執(zhí)行指定的入口類。命令執(zhí)行后可以訪問(wèn)對(duì)應(yīng)的Flink WebUI:https://node1:8081，可以看到提交的任務(wù)，但是由于還沒(méi)有執(zhí)行TaskManager任務(wù)無(wú)法執(zhí)行。
1.3、啟動(dòng)TaskManager
在node1、node2、node3任意一臺(tái)節(jié)點(diǎn)上啟動(dòng)taskManager，根據(jù)$FLINK_HOME/conf/flink-conf.yaml配置文件中"jobmanager.rpc.address"配置項(xiàng)會(huì)找到對(duì)應(yīng)node1 JobManager。
#在node1節(jié)點(diǎn)上啟動(dòng)TaskManager[root@node1 ~]# cd /software/flink-1.16.0/bin/[root@node1 bin]# ./taskmanager.sh start#在node2節(jié)點(diǎn)上啟動(dòng)TaskManager[root@node2 ~]# cd /software/flink-1.16.0/bin/[root@node2 bin]# ./taskmanager.sh start啟動(dòng)兩個(gè)TaskManager后可以看到Flink WebUI中對(duì)應(yīng)的有2個(gè)TaskManager，可以根據(jù)自己任務(wù)使用資源的情況，手動(dòng)啟動(dòng)多個(gè)TaskManager。
1.4、停止集群
#停止啟動(dòng)的JobManager[root@node1 bin]# ./standalone-job.sh stop#停止啟動(dòng)的TaskManager[root@node1 bin]# ./taskmanager.sh stop[root@node2 bin]# ./taskmanager.sh stop我們可以以同樣的方式在其他節(jié)點(diǎn)上以Standalone Application模式提交先的Flink任務(wù)，但是每次提交都是當(dāng)前提交任務(wù)獨(dú)享集群資源。
Standalone Application模式提交任務(wù)中提交任務(wù)的同時(shí)會(huì)啟動(dòng)JobManager創(chuàng)建Flink集群，但是需要手動(dòng)啟動(dòng)TaskManager，這樣提交的任務(wù)才能正常運(yùn)行，如果提交的任務(wù)使用資源多，還可以啟動(dòng)多個(gè)TaskManager。Standalone Application模式提交任務(wù)流程如下：
在客戶端提交Flink任務(wù)的同時(shí)啟動(dòng)JobManager，客戶端會(huì)將任務(wù)轉(zhuǎn)換成JobGraph提交給JobManager。Dispatcher會(huì)啟動(dòng)JobMaster，Dispatcher將提交任務(wù)提交給JobMaster。JobMaster向ResourceManager申請(qǐng)Slot資源。手動(dòng)啟動(dòng)TaskManager，TaskManager會(huì)向ResourceManager注冊(cè)Slot資源ResourceManager會(huì)在對(duì)應(yīng)的TaskManager上劃分Slot資源。TaskManager向JobMaster offer Slot資源。JobMaster將任務(wù)對(duì)應(yīng)的task發(fā)送到TaskManager上執(zhí)行。Standalone Application模式任務(wù)提交流程和Standalone Session模式類似，兩者區(qū)別主要是Standalone Session模式中啟動(dòng)Flink集群時(shí)JobManager、TaskManager、JobMaster會(huì)預(yù)先啟動(dòng)；Standalone Application模式中提交任務(wù)時(shí)同時(shí)啟動(dòng)集群JobManager、JobMaster，需要手動(dòng)啟動(dòng)TaskManager。
?
上一篇：寧波銀行上海分行獲評(píng)人行上海分行2022年上海市銀行業(yè)現(xiàn)金服務(wù)與管理業(yè)務(wù)考核A類行
責(zé)任編輯：
