《數(shù)據(jù)實(shí)時(shí)處理flink》課件-第九章 Flink的部署和配置

Flink的部署和配置通過(guò)對(duì)前文的學(xué)習(xí)，包括使用DataStreamAPI和使用TableAPI&SQL來(lái)編寫程序。本章將重點(diǎn)介紹如何部署和配置Flink作業(yè)。我們將分別介紹Standalone、Yarn、Kubernetes上部署Flink集群。以及一些配置細(xì)節(jié)。通過(guò)本節(jié)學(xué)習(xí)您將可以：掌握Flink集群幾種常見的部署模式。了解如何配置一個(gè)Flink作業(yè)，包括CPU、內(nèi)存和硬盤。了解如何設(shè)置算子鏈與槽位共享。掌握如何使用命令行工具提交和管理作業(yè)。了解如何與Hadoop集成。Flink集群部署模式Flink配置文件命令行工具Flink與Hadoop集成

部署模式Standalone集群：集群內(nèi)只部署FlinkHadoop

YARN集群：兼容Hadoop生態(tài)，可以部署Hadoop、Hive、Spark、Flink…Kubernetes集群：兼容各類容器至少一個(gè)Master，至少一個(gè)WorkerMaster：Dispatcher、ResourceManager和JobManagerWorker：TaskManager安裝好Java、配置好SSH免密碼訪問(wèn)配置Flink主目錄的conf/flink-conf.yaml和conf/slaves兩個(gè)文件將Flink主目錄部署到每個(gè)節(jié)點(diǎn)的相同路徑，或者部署到所有節(jié)點(diǎn)都可訪問(wèn)的共享目錄啟動(dòng)這個(gè)集群：bin/start-cluster.sh

Standalone集群一個(gè)Standalone集群拓?fù)錁永?./bin/flinkrun-m:8081./examples/batch/WordCount.jar提交作業(yè)：使用針對(duì)Hadoop的Flink，設(shè)置HADOOP_CONF_DIR和YARN_CONF_DIR

等與Hadoop相關(guān)的配置啟動(dòng)好YARN集群三種作業(yè)提交方式：Per-Job：每次向YARN提交一個(gè)作業(yè)，YARN為這個(gè)作業(yè)單獨(dú)分配資源，基于這些資源啟動(dòng)一個(gè)Flink集群，作業(yè)運(yùn)行結(jié)束，相應(yīng)的資源會(huì)被釋放。Session：在YARN上啟動(dòng)一個(gè)長(zhǎng)期運(yùn)行的Flink集群，用戶可以向這個(gè)集群提交多個(gè)作業(yè)。Application：在Per-Job模式上做了一些優(yōu)化。Hadoop

YARNClient將作業(yè)提交給YARN的ResourceManagerYARN為這個(gè)作業(yè)生成一個(gè)Application

Master以運(yùn)行FlinkMaster，里面運(yùn)行著JobManager和Flink-YARNResourceManagerJobManager會(huì)根據(jù)本次作業(yè)所需資源向Flink-YARNResourceManager申請(qǐng)Slot資源Flink-YARNResourceManager會(huì)向YARN申請(qǐng)所需的資源作為TaskManagerTaskManager將這些Slot注冊(cè)到Flink-

YARNResourceManagerJobManager將作業(yè)的計(jì)算任務(wù)部署到各TaskManager上YARN

Per-JobYARN

Per-Job模式在Client上，使用bin/yarn-session.sh啟動(dòng)一個(gè)YARNSession，F(xiàn)link向YARNResourceManager申請(qǐng)一個(gè)Application

Master用戶提交一個(gè)作業(yè)，作業(yè)被發(fā)送給DispatcherDispatcher會(huì)啟動(dòng)針對(duì)該作業(yè)的JobManagerJobManager向Flink-YARN

ResourceManager上申請(qǐng)所需資源，啟動(dòng)TaskManagerTaskManager注冊(cè)JobManager將計(jì)算任務(wù)部署到各TaskManager上如果用戶提交下一個(gè)作業(yè)，那么Dispatcher啟動(dòng)新的JobManager，新的JobManager負(fù)責(zé)新作業(yè)的資源申請(qǐng)和任務(wù)調(diào)度。YARN

SessionYARN

Session模式Per-Job和Session模式作業(yè)提交的過(guò)程依賴Client，main()方法在Client上執(zhí)行，main()方法會(huì)將作業(yè)的各個(gè)依賴下載到本地，生成JobGraph，并將依賴以及JobGraph發(fā)送到Flink集群,負(fù)載很重。Application模式允許main()方法在JobManager上執(zhí)行，這樣可以分擔(dān)Client的壓力YARN

Application兩種方式：Kubernetes和Kubernetes原生Kubernetes原生Session模式：使用bin/kubernetes-session.sh啟動(dòng)一個(gè)KubernetesSessionKubernetes相關(guān)組件將進(jìn)行初始化，生成Flink

Master（Dispatcher、Flink-Kubernetes

ResourceManager）用戶提交作業(yè)，申請(qǐng)資源，將作業(yè)運(yùn)行到TaskManager上Kubernetes原生Application模式KubernetesKubernetes原生Session模式Flink集群部署模式Flink配置文件命令行工具Flink與Hadoop集成使用Key-Value來(lái)設(shè)置參數(shù)，很多進(jìn)程會(huì)讀取這個(gè)文件從Flink官網(wǎng)下載的Flink主目錄里的flink-conf.yaml有一些默認(rèn)設(shè)置，針對(duì)單機(jī)環(huán)境，在自己環(huán)境中使用，需要修改這個(gè)文件conf/flink-conf.yaml

安裝Java，設(shè)置$JAVA_HOME類加載：將Java的.class文件加載到JVM虛擬機(jī)一個(gè)Flink作業(yè)主要加載兩種類：JavaClasspath：JDK核心類庫(kù)和Flink主目錄下lib文件夾中的類用戶類（UserCode）：用戶編寫的應(yīng)用作業(yè)中的類類加載策略：子類優(yōu)先（Child-first）：優(yōu)先加載用戶編寫的應(yīng)用作業(yè)中的類，F(xiàn)link默認(rèn)使用Child-first父類優(yōu)先（Parent-first）：優(yōu)先加載JavaClasspath中的類Java和類加載并行度（parallelism

）：每個(gè)算子都會(huì)被切分為parallelism個(gè)子任務(wù)，分布到多個(gè)Slot上。根據(jù)計(jì)算規(guī)模大小，調(diào)整并行度。如果作業(yè)開啟了算子鏈和槽位共享，一個(gè)Slot上運(yùn)行著一個(gè)作業(yè)所有算子組成的流水線（Pipeline），這個(gè)作業(yè)需要parallelism個(gè)Slot

。槽位劃分：將TaskManager劃分成多少個(gè)Slot。使用taskmanager.numberOfTaskSlots

設(shè)置默認(rèn)值為1，Standalone集群官方建議將參數(shù)值配置為與CPU核心數(shù)相等或成比例配置taskmanager.numberOfTaskSlots沒(méi)有絕對(duì)的準(zhǔn)則：每個(gè)TaskManager下有一個(gè)Slot，該Slot會(huì)獨(dú)立運(yùn)行在一個(gè)JVM進(jìn)程中。隔離性好。每個(gè)TaskManager下有多個(gè)Slot，那么多個(gè)Slot同時(shí)運(yùn)行在一個(gè)JVM進(jìn)程中。多個(gè)Slot可以共享資源，隔離性差。并行度和槽位劃分堆區(qū)（Heap）:JVM虛擬化之后的內(nèi)存，存儲(chǔ)Java對(duì)象實(shí)例，使用Java垃圾回收（GarbageCollection，GC）機(jī)制來(lái)清理內(nèi)存中的不再使用的對(duì)象。某個(gè)時(shí)間點(diǎn)，必須進(jìn)行一次Full

GC，F(xiàn)ull

GC影響Java應(yīng)用的性能。悖論：Heap越大，F(xiàn)ull

GC時(shí)間越長(zhǎng)。Heap太小，會(huì)出現(xiàn)OutOfMemoryError異常。堆外（Off-Heap）：直接由操作系統(tǒng)管理的內(nèi)存，適合讀寫操作頻繁的場(chǎng)景。使用、監(jiān)控和調(diào)試更復(fù)雜。Full

GC影響性能時(shí)，可以考慮Off-Heap。Java內(nèi)存JVM進(jìn)程：Master、TaskManager…Flink占用的內(nèi)存（TotalFlinkMemory）：Flink可以使用的內(nèi)存JVMHeapOff-heapMemoryFlinkManagedMemory：TaskManager個(gè)別場(chǎng)景使用DirectMemory

：網(wǎng)絡(luò)通信緩存使用的內(nèi)存JVM相關(guān)內(nèi)存（JVMSpecificMemory）：Java程序都需要的一塊內(nèi)存區(qū)域Flink將內(nèi)存管理部分做了封裝，用戶在絕大多數(shù)情況下不用關(guān)注數(shù)據(jù)到底是如何寫入內(nèi)存的Flink內(nèi)存模型Master占用內(nèi)存不大最簡(jiǎn)單的配置方法：設(shè)置TotalProcessMemory（cess.size）

，根據(jù)默認(rèn)分配比例，

將內(nèi)存分配給各個(gè)模塊。設(shè)置TotalFlinkMemory（jobmanager.memory.flink.size），如果Master進(jìn)程需要管理多個(gè)作業(yè)，

需要增大這個(gè)配置。Master的內(nèi)存配置TaskManager涉及數(shù)據(jù)處理，對(duì)內(nèi)存的需求很高根據(jù)是否為Flink框架所用，分為Flink框架和非Flink框架，框架使用的內(nèi)存，用戶任務(wù)無(wú)法使用（Framework

Heap，F(xiàn)ramework

Off-heap）用戶任務(wù)使用的內(nèi)存（Task

Heap、Task

Off-heap、Flink

Managed

Memory和Network）根據(jù)堆區(qū)非堆區(qū)，分為Heap和Off-heapTask

Heap：用戶程序內(nèi)存FlinkManagedMemory：流處理下RocksDB的StateBackend，批處理下排序、中間數(shù)據(jù)緩存等Network：Netty緩存TaskManager的內(nèi)存配置粗粒度的配置方法：配置TotalProcessMemory或TotalFlinkMemory兩者中的任意一個(gè)。各個(gè)子模塊根據(jù)默認(rèn)比例獲得相應(yīng)的內(nèi)存。Total

Process

Memory：整個(gè)進(jìn)程的內(nèi)存，適合容器化部署方式。（cess.size）Total

Flink

Memory：Flink可用內(nèi)存，適合Standalone集群方式。（taskmanager.memory.flink.size）細(xì)粒度的配置方法同時(shí)配置TaskHeap和FlinkManagedMemory兩個(gè)內(nèi)存。這兩個(gè)內(nèi)存服務(wù)于具體的計(jì)算任務(wù)。（taskmanager.memory.task.heap.size和taskmanager.memory.managed.size）以上3個(gè)參數(shù)不要同時(shí)配置，否則會(huì)引起沖突。TaskManager的內(nèi)存配置Flink會(huì)將一部分?jǐn)?shù)據(jù)寫到本地磁盤，比如：日志信息、RocksDB數(shù)據(jù)等。io.tmp.dirs配置了本地磁盤讀寫位置，默認(rèn)會(huì)使用JVM的參數(shù)java.io.tmpdir

Linux下一般為/tmp磁盤Flink使用算子鏈將盡可能多的上、下游算子鏈接到一起。上、下游算子會(huì)被捆綁到一起，作為一個(gè)線程執(zhí)行?？梢蕴岣哔Y源利用率。如果兩個(gè)算子不做算子鏈，算子間數(shù)據(jù)通信存在序列化和反序列化，通信成本高。數(shù)據(jù)不發(fā)生交換，才可以進(jìn)行算子鏈。兩個(gè)算子負(fù)載都很高，不應(yīng)該進(jìn)行算子鏈。默認(rèn)開啟算子鏈。env.disableOperatorChaining()：關(guān)閉算子鏈。startNewChain()：對(duì)特定算子開啟新的算子鏈。算子鏈和槽位共享Flink默認(rèn)開啟了槽位共享：從Source到Sink的所有算子子任務(wù)組成的Pipeline可以共享一個(gè)Slot。也可以手動(dòng)設(shè)置槽位共享組（SlotSharingGroup）。槽位共享整個(gè)作業(yè)并行度為2，從Source到Sink所有算子共享一個(gè)Slot。給WindowAggreagtion設(shè)置SlotSharingGroup，該算子及之后的算子被劃分到另外的Slot

stream.timeWindow(...).sum(...).slotSharingGroup(“A”);Flink集群部署模式Flink配置文件命令行工具Flink與Hadoop集成生產(chǎn)環(huán)境中，一般使用命令行管理作業(yè)，名為flink，放在Flink主目錄下的bin目錄下。功能：提交、取消、羅列當(dāng)前作業(yè)，獲取作業(yè)信息，設(shè)置Savepoint。會(huì)從conf/flink-conf.yaml里讀取配置信息。使用方式：ACTION包括run（提交作業(yè)）、stop（取消作業(yè)）等。OPTIONS為一些預(yù)置的選項(xiàng)，ARGUMENTS是用戶傳入的參數(shù)。命令行工具$

./bin/flink<ACTION>[OPTIONS][ARGUMENTS]模板：需要使用Maven對(duì)用戶代碼打包，得到JAR包。使用WordCount的例子：使用-c參數(shù)，設(shè)置程序入口類：在命令行中用-p選項(xiàng)設(shè)置這個(gè)作業(yè)的并行度提交作業(yè)$

./bin/flinkrun[OPTIONS]<xxx.jar>[ARGUMENTS]$

./bin/flinkrun./examples/streaming/WordCount.jar$

./bin/flinkrun./examples/streaming/WordCount.jar -corg.apache.flink.streaming.examples.wordcount.WordCount$

./bin/flinkrun-p2./examples/streaming/WordCount.jar傳入?yún)?shù)，參數(shù)會(huì)寫入main方法的參數(shù)String[]中使用-m選項(xiàng)來(lái)設(shè)置向具體哪個(gè)Master提交提交作業(yè)$

./bin/flinkrun./examples/streaming/WordCount.jar -corg.apache.flink.streaming.examples.wordcount.WordCount --input'/tmp/a.log’

--output'/tmp/b.log'

$

./bin/flinkrun-mmyJMHost:8081 \./examples/streaming/WordCount.jar羅列當(dāng)前作業(yè)：觸發(fā)一個(gè)作業(yè)執(zhí)行Savepoint，savepointDirectory為目錄：關(guān)停一個(gè)Flink作業(yè)：從一個(gè)Savepoint恢復(fù)一個(gè)應(yīng)用作業(yè)：管理作業(yè)$

./bin/flinklist$

./bin/flinksavepoint<jobId>[savepointDirectory]$

./bin/flinkcancel<jobID>$

./bin/flinkrun-s<savepointPath>[OPTIONS]<xxx.jar>Flink集群部署模式Fli