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

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

部署模式Standalone集群：集群內只部署FlinkHadoop

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

Standalone集群一個Standalone集群拓撲樣例$./bin/flinkrun-m:8081./examples/batch/WordCount.jar提交作業(yè)：使用針對Hadoop的Flink，設置HADOOP_CONF_DIR和YARN_CONF_DIR

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

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

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

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

Per-JobYARN

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

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

ResourceManager上申請所需資源，啟動TaskManagerTaskManager注冊JobManager將計算任務部署到各TaskManager上如果用戶提交下一個作業(yè)，那么Dispatcher啟動新的JobManager，新的JobManager負責新作業(yè)的資源申請和任務調度。YARN

SessionYARN

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

Application兩種方式：Kubernetes和Kubernetes原生Kubernetes原生Session模式：使用bin/kubernetes-session.sh啟動一個KubernetesSessionKubernetes相關組件將進行初始化，生成Flink

Master（Dispatcher、Flink-Kubernetes

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

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

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

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

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

GC，F(xiàn)ull

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

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

GC影響性能時，可以考慮Off-Heap。Java內存JVM進程：Master、TaskManager…Flink占用的內存（TotalFlinkMemory）：Flink可以使用的內存JVMHeapOff-heapMemoryFlinkManagedMemory：TaskManager個別場景使用DirectMemory

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

，根據默認分配比例，

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

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

Heap，F(xiàn)ramework

Off-heap）用戶任務使用的內存（Task

Heap、Task

Off-heap、Flink

Managed

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

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

Process

Memory：整個進程的內存，適合容器化部署方式。（cess.size）Total

Flink

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

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

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

./bin/flink<ACTION>[OPTIONS][ARGUMENTS]模板：需要使用Maven對用戶代碼打包，得到JAR包。使用WordCount的例子：使用-c參數(shù)，設置程序入口類：在命令行中用-p選項設置這個作業(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傳入參數(shù)，參數(shù)會寫入main方法的參數(shù)String[]中使用-m選項來設置向具體哪個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羅列當前作業(yè)：觸發(fā)一個作業(yè)執(zhí)行Savepoint，savepointDirectory為目錄：關停一個Flink作業(yè)：從一個Savepoint恢復一個應用作業(yè)：管理作業(yè)$

./bin/flinklist$

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

./bin/flinkcancel<jobID>$

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