如何實時監(jiān)控 Flink 集群和作業(yè)？
Flink 相關(guān)的組件和作業(yè)的穩(wěn)定性通常是比較關(guān)鍵的，所以得需要對它們進行監(jiān)控，如果有異常，則需要及時告警通知。本章先會教會教會大家如何利用現(xiàn)有 Flink UI 上面的信息去發(fā)現(xiàn)和排查問題，會指明一些比較重要和我們非常關(guān)心的指標，通過這些指標我們能夠立馬定位到問題的根本原因。接著筆者會教大家如何去利用現(xiàn)有的 Metrics Reporter 去構(gòu)建一個 Flink 的監(jiān)控系統(tǒng)，它可以收集到所有作業(yè)的監(jiān)控指標，并會存儲這些監(jiān)控指標數(shù)據(jù)，最后還會有一個監(jiān)控大盤做數(shù)據(jù)可視化，通過這個大盤可以方便排查問題。
實時監(jiān)控 Flink 及其作業(yè)
當將 Flink JobManager、TaskManager 都運行起來了，并且也部署了不少 Flink Job，那么它到底是否還在運行、運行的狀態(tài)如何、資源 TaskManager 和 Slot 的個數(shù)是否足夠、Job 內(nèi)部是否出現(xiàn)異常、計算速度是否跟得上數(shù)據(jù)生產(chǎn)的速度 等這些問題其實對我們來說是比較關(guān)注的，所以就很迫切的需要一個監(jiān)控系統(tǒng)幫我們把整個 Flink 集群的運行狀態(tài)給展示出來。通過監(jiān)控系統(tǒng)我們能夠很好的知道 Flink 內(nèi)部的整個運行狀態(tài)，然后才能夠根據(jù)項目生產(chǎn)環(huán)境遇到的問題 ‘對癥下藥’。下面分別來講下 JobManager、TaskManager、Flink Job 的監(jiān)控以及最關(guān)心的一些監(jiān)控指標。
?監(jiān)控 JobManager
我們知道 JobManager 是 Flink 集群的中控節(jié)點，類似于 Apache Storm 的 Nimbus 以及 Apache Spark 的 Driver 的角色。它負責作業(yè)的調(diào)度、作業(yè) Jar 包的管理、Checkpoint 的協(xié)調(diào)和發(fā)起、與 TaskManager 之間的心跳檢查等工作。如果 JobManager 出現(xiàn)問題的話，就會導(dǎo)致作業(yè) UI 信息查看不了，TaskManager 和所有運行的作業(yè)都會受到一定的影響，所以這也是為啥在 7.1 節(jié)中強調(diào) JobManager 的高可用問題。
在 Flink 自帶的 UI 上 JobManager 那個 Tab 展示的其實并沒有顯示其對應(yīng)的 Metrics，那么對于 JobManager 來說常見比較關(guān)心的監(jiān)控指標有哪些呢？
基礎(chǔ)指標
因為 Flink JobManager 其實也是一個 Java 的應(yīng)用程序，那么它自然也會有 Java 應(yīng)用程序的指標，比如內(nèi)存、CPU、GC、類加載、線程信息等。
?內(nèi)存：內(nèi)存又分堆內(nèi)存和非堆內(nèi)存，在 Flink 中還有 Direct 內(nèi)存，每種內(nèi)存又有初始值、使用值、最大值等指標，因為在 JobManager 中的工作其實相當于 TaskManager 來說比較少，也不存儲事件數(shù)據(jù)，所以通常 JobManager 占用的內(nèi)存不會很多，在 Flink JobManager 中自帶的內(nèi)存 Metrics 指標有：
jobmanager_Status_JVM_Memory_Direct_Countjobmanager_Status_JVM_Memory_Direct_MemoryUsedjobmanager_Status_JVM_Memory_Direct_TotalCapacityjobmanager_Status_JVM_Memory_Heap_Committedjobmanager_Status_JVM_Memory_Heap_Maxjobmanager_Status_JVM_Memory_Heap_Usedjobmanager_Status_JVM_Memory_Mapped_Countjobmanager_Status_JVM_Memory_Mapped_MemoryUsedjobmanager_Status_JVM_Memory_Mapped_TotalCapacityjobmanager_Status_JVM_Memory_NonHeap_Committedjobmanager_Status_JVM_Memory_NonHeap_Maxjobmanager_Status_JVM_Memory_NonHeap_Used
?CPU：JobManager 分配的 CPU 使用情況，如果使用類似 K8S 等資源調(diào)度系統(tǒng)，則需要對每個容器進行設(shè)置資源，比如 CPU 限制不能超過多少，在 Flink JobManager 中自帶的 CPU 指標有：
jobmanager_Status_JVM_CPU_Loadjobmanager_Status_JVM_CPU_Time
?GC：GC 信息對于 Java 應(yīng)用來說是避免不了的，每種 GC 都有時間和次數(shù)的指標可以供參考，提供的指標有：
jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Countjobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Timejobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Countjobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Time
Checkpoint 指標
因為 JobManager 負責了作業(yè)的 Checkpoint 的協(xié)調(diào)和發(fā)起功能，所以 Checkpoint 相關(guān)的指標就有表示 Checkpoint 執(zhí)行的時間、Checkpoint 的時間長短、完成的 Checkpoint 的次數(shù)、Checkpoint 失敗的次數(shù)、Checkpoint 正在執(zhí)行 Checkpoint 的個數(shù)等，其對應(yīng)的指標如下：
jobmanager_job_lastCheckpointAlignmentBufferedjobmanager_job_lastCheckpointDurationjobmanager_job_lastCheckpointExternalPathjobmanager_job_lastCheckpointRestoreTimestampjobmanager_job_lastCheckpointSizejobmanager_job_numberOfCompletedCheckpointsjobmanager_job_numberOfFailedCheckpointsjobmanager_job_numberOfInProgressCheckpointsjobmanager_job_totalNumberOfCheckpoints
重要的指標
另外還有比較重要的指標就是 Flink UI 上也提供的，類似于 Slot 總共個數(shù)、Slot 可使用的個數(shù)、TaskManager 的個數(shù)（通過查看該值可以知道是否有 TaskManager 發(fā)生異常重啟）、正在運行的作業(yè)數(shù)量、作業(yè)運行的時間和完成的時間、作業(yè)的重啟次數(shù)，對應(yīng)的指標如下：
jobmanager_job_uptimejobmanager_numRegisteredTaskManagersjobmanager_numRunningJobsjobmanager_taskSlotsAvailablejobmanager_taskSlotsTotaljobmanager_job_downtimejobmanager_job_fullRestartsjobmanager_job_restartingTime
監(jiān)控 TaskManager
TaskManager 在 Flink 集群中也是一個個的進程實例，它的數(shù)量代表著能夠運行作業(yè)個數(shù)的能力，所有的 Flink 作業(yè)最終其實是會在 TaskManager 上運行的，TaskManager 管理著運行在它上面的所有作業(yè)的 Task 的整個生命周期，包括了 Task 的啟動銷毀、內(nèi)存管理、磁盤 IO、網(wǎng)絡(luò)傳輸管理等。
因為所有的 Task 都是運行運行在 TaskManager 上的，有的 Task 可能會做比較復(fù)雜的操作或者會存儲很多數(shù)據(jù)在內(nèi)存中，那么就會消耗很大的資源，所以通常來說 TaskManager 要比 JobManager 消耗的資源要多，但是這個資源具體多少其實也不好預(yù)估，所以可能會出現(xiàn)由于分配資源的不合理，導(dǎo)致 TaskManager 出現(xiàn) OOM 等問題。一旦 TaskManager 因為各種問題導(dǎo)致崩潰重啟的話，運行在它上面的 Task 也都會失敗，JobManager 與它的通信也會丟失。因為作業(yè)出現(xiàn) failover，所以在重啟這段時間它是不會去消費數(shù)據(jù)的，所以必然就會出現(xiàn)數(shù)據(jù)消費延遲的問題。對于這種情況那么必然就很需要 TaskManager 的監(jiān)控信息，這樣才能夠?qū)φ麄€集群的 TaskManager 做一個提前預(yù)警。
那么在 Flink 中自帶的 TaskManager Metrics 有哪些呢？主要也是 CPU、類加載、GC、內(nèi)存、網(wǎng)絡(luò)等。其實這些信息在 Flink UI 上也是有，如下圖所示，不知道讀者有沒有細心觀察過。
在這個 TaskManager 的 Metrics 監(jiān)控頁面通常比較關(guān)心的指標有內(nèi)存相關(guān)的，還有就是 GC 的指標，通常一個 TaskManager 出現(xiàn) OOM 之前會不斷的進行 GC，在這個 Metrics 頁面它展示了年輕代和老年代的 GC 信息（時間和次數(shù)），如下圖所示，大家可以細心觀察下是否 TaskManager OOM 前老年代和新生代的 GC 次數(shù)比較、時間比較長。
在 Flink Reporter 中提供的 TaskManager Metrics 指標如下：
taskmanager_Status_JVM_CPU_Loadtaskmanager_Status_JVM_CPU_Timetaskmanager_Status_JVM_ClassLoader_ClassesLoadedtaskmanager_Status_JVM_ClassLoader_ClassesUnloadedtaskmanager_Status_JVM_GarbageCollector_G1_Old_Generation_Counttaskmanager_Status_JVM_GarbageCollector_G1_Old_Generation_Timetaskmanager_Status_JVM_GarbageCollector_G1_Young_Generation_Counttaskmanager_Status_JVM_GarbageCollector_G1_Young_Generation_Timetaskmanager_Status_JVM_Memory_Direct_Counttaskmanager_Status_JVM_Memory_Direct_MemoryUsedtaskmanager_Status_JVM_Memory_Direct_TotalCapacitytaskmanager_Status_JVM_Memory_Heap_Committedtaskmanager_Status_JVM_Memory_Heap_Maxtaskmanager_Status_JVM_Memory_Heap_Usedtaskmanager_Status_JVM_Memory_Mapped_Counttaskmanager_Status_JVM_Memory_Mapped_MemoryUsedtaskmanager_Status_JVM_Memory_Mapped_TotalCapacitytaskmanager_Status_JVM_Memory_NonHeap_Committedtaskmanager_Status_JVM_Memory_NonHeap_Maxtaskmanager_Status_JVM_Memory_NonHeap_Usedtaskmanager_Status_JVM_Threads_Counttaskmanager_Status_Network_AvailableMemorySegmentstaskmanager_Status_Network_TotalMemorySegmentstaskmanager_Status_Shuffle_Netty_AvailableMemorySegmentstaskmanager_Status_Shuffle_Netty_TotalMemorySegments
監(jiān)控 Flink 作業(yè)
對于運行的作業(yè)來說，其實我們會更關(guān)心其運行狀態(tài)，如果沒有其對應(yīng)的一些監(jiān)控信息，那么對于我們來說這個 Job 就是一個黑盒，完全不知道是否在運行，Job 運行狀態(tài)是什么、Task 運行狀態(tài)是什么、是否在消費數(shù)據(jù)、消費數(shù)據(jù)是咋樣（細分到每個 Task）、消費速度能否跟上生產(chǎn)數(shù)據(jù)的速度、處理數(shù)據(jù)的過程中是否有遇到什么錯誤日志、處理數(shù)據(jù)是否有出現(xiàn)反壓問題等等。
上面列舉的這些問題通常來說是比較關(guān)心的，那么在 Flink UI 上也是有提供的查看對應(yīng)的信息的，點開對應(yīng)的作業(yè)就可以查看到作業(yè)的執(zhí)行圖，每個 Task 的信息都是會展示出來的，包含了狀態(tài)、Bytes Received（接收到記錄的容量大小）、Records Received（接收到記錄的條數(shù)）、Bytes Sent（發(fā)出去的記錄的容量大小）、Records Sent（發(fā)出去記錄的條數(shù)）、異常信息、timeline（作業(yè)運行狀態(tài)的時間線）、Checkpoint 信息，如下圖所示。
這些指標也可以通過 Flink 的 Reporter 進行上報存儲到第三方的時序數(shù)據(jù)庫，然后通過類似 Grafana 展示出來，如下圖所示。通過這些信息大概就可以清楚的知道一個 Job 的整個運行狀態(tài)，然后根據(jù)這些運行狀態(tài)去分析作業(yè)是否有問題。
在流作業(yè)中最關(guān)鍵的指標無非是作業(yè)的實時性，那么延遲就是衡量作業(yè)的是否實時的一個基本參數(shù)，但是對于現(xiàn)有的這些信息其實還不知道作業(yè)的消費是否有延遲，通常來說可以結(jié)合 Kafka 的監(jiān)控去查看對應(yīng)消費的 Topic 的 Group 的 Lag 信息，如果 Lag 很大就表明有數(shù)據(jù)堆積了，另外還有一個辦法就是需要自己在作業(yè)中自定義 Metrics 做埋點，將算子在處理數(shù)據(jù)的系統(tǒng)時間與數(shù)據(jù)自身的 Event Time 做一個差值，求得值就可以知道算子消費的數(shù)據(jù)是什么時候的了。比如在 1571457964000（2019-10-19 12:06:04）Map 算子消費的數(shù)據(jù)的事件時間是 1571457604000（2019-10-19 12:00:04），相差了 6 分鐘，那么就表明消費延遲了 6 分鐘，然后通過 Metrics Reporter 將埋點的 Metrics 信息上傳，這樣最終就可以獲取到作業(yè)在每個算子處的消費延遲的時間。
上面的是針對于作業(yè)延遲的判斷方法，另外像類似于作業(yè)反壓的情況，在 Flink 的 UI 也會有展示，具體怎么去分析和處理這種問題在 9.1 節(jié)中有詳細講解。
根據(jù)這些監(jiān)控信息不僅可以做到提前預(yù)警，做好資源的擴容（比如增加容器的數(shù)量／內(nèi)存／CPU／并行度／Slot 個數(shù)），也還可以找出作業(yè)配置的資源是否有浪費。通常來說一個作業(yè)的上線可能是會經(jīng)過資源的預(yù)估，然后才會去申請這個作業(yè)要配置多少資源，比如算子要使用多少并行度，最后上線后可以通過完整的運行監(jiān)控信息查看該作業(yè)配置的并行度是否有過多或者配置的內(nèi)存比較大。比如出現(xiàn)下面這些情況的時候可能就是資源出現(xiàn)浪費了：
?作業(yè)消費從未發(fā)生過延遲，即使在數(shù)據(jù)流量高峰的時候，也未發(fā)生過消費延遲?作業(yè)運行所在的 TaskManager 堆內(nèi)存使用率異常的低?作業(yè)運行所在的 TaskManager 的 GC 時間和次數(shù)非常規(guī)律，沒有出現(xiàn)異常的現(xiàn)象，如下圖所示。
在 Flink Metrics Reporter 上傳的指標中大概有下面這些：
taskmanager_job_task_Shuffle_Netty_Input_Buffers_outPoolUsagetaskmanager_job_task_Shuffle_Netty_Input_Buffers_outputQueueLengthtaskmanager_job_task_Shuffle_Netty_Output_Buffers_inPoolUsagetaskmanager_job_task_Shuffle_Netty_Output_Buffers_inputExclusiveBuffersUsagetaskmanager_job_task_Shuffle_Netty_Output_Buffers_inputFloatingBuffersUsagetaskmanager_job_task_Shuffle_Netty_Output_Buffers_inputQueueLengthtaskmanager_job_task_Shuffle_Netty_Output_numBuffersInLocaltaskmanager_job_task_Shuffle_Netty_Output_numBuffersInLocalPerSecondtaskmanager_job_task_Shuffle_Netty_Output_numBuffersInRemotetaskmanager_job_task_Shuffle_Netty_Output_numBuffersInRemotePerSecondtaskmanager_job_task_Shuffle_Netty_Output_numBytesInLocaltaskmanager_job_task_Shuffle_Netty_Output_numBytesInLocalPerSecondtaskmanager_job_task_Shuffle_Netty_Output_numBytesInRemotetaskmanager_job_task_Shuffle_Netty_Output_numBytesInRemotePerSecondtaskmanager_job_task_buffers_inPoolUsagetaskmanager_job_task_buffers_inputExclusiveBuffersUsagetaskmanager_job_task_buffers_inputFloatingBuffersUsagetaskmanager_job_task_buffers_inputQueueLengthtaskmanager_job_task_buffers_outPoolUsagetaskmanager_job_task_buffers_outputQueueLengthtaskmanager_job_task_checkpointAlignmentTimetaskmanager_job_task_currentInputWatermarktaskmanager_job_task_numBuffersInLocaltaskmanager_job_task_numBuffersInLocalPerSecondtaskmanager_job_task_numBuffersInRemotetaskmanager_job_task_numBuffersInRemotePerSecondtaskmanager_job_task_numBuffersOuttaskmanager_job_task_numBuffersOutPerSecondtaskmanager_job_task_numBytesIntaskmanager_job_task_numBytesInLocaltaskmanager_job_task_numBytesInLocalPerSecondtaskmanager_job_task_numBytesInPerSecondtaskmanager_job_task_numBytesInRemotetaskmanager_job_task_numBytesInRemotePerSecondtaskmanager_job_task_numBytesOuttaskmanager_job_task_numBytesOutPerSecondtaskmanager_job_task_numRecordsIntaskmanager_job_task_numRecordsInPerSecondtaskmanager_job_task_numRecordsOuttaskmanager_job_task_numRecordsOutPerSecondtaskmanager_job_task_operator_currentInputWatermarktaskmanager_job_task_operator_currentOutputWatermarktaskmanager_job_task_operator_numLateRecordsDroppedtaskmanager_job_task_operator_numRecordsIntaskmanager_job_task_operator_numRecordsInPerSecondtaskmanager_job_task_operator_numRecordsOuttaskmanager_job_task_operator_numRecordsOutPerSecond
最關(guān)心的性能指標
上面已經(jīng)提及到 Flink 的 JobManager、TaskManager 和運行的 Flink Job 的監(jiān)控以及常用的監(jiān)控信息，這些指標有的是可以直接在 Flink 的 UI 上觀察到的，另外 Flink 提供了 Metrics Reporter 進行上報存儲到監(jiān)控系統(tǒng)中去，然后通過可視化的圖表進行展示，在 8.2 節(jié)中將教大家如何構(gòu)建一個完整的監(jiān)控系統(tǒng)。那么有了這么多監(jiān)控指標，其實哪些是比較重要的呢，比如說這些指標出現(xiàn)異常的時候可以發(fā)出告警及時進行通知，這樣可以做到預(yù)警作用，另外還可以根據(jù)這些信息進行作業(yè)資源的評估。下面列舉一些筆者覺得比較重要的指標：
JobManager
在 JobManager 中有著該集群中所有的 TaskManager 的個數(shù)、Slot 的總個數(shù)、Slot 的可用個數(shù)、運行的時間、作業(yè)的 Checkpoint 情況，筆者覺得這幾個指標可以重點關(guān)注。
?TaskManager 個數(shù)：如果出現(xiàn) TaskManager 突然減少，可能是因為有 TaskManager 掛掉重啟，一旦該 TaskManager 之前運行了很多作業(yè)，那么重啟帶來的影響必然是巨大的。?Slot 個數(shù)：取決于 TaskManager 的個數(shù)，決定了能運行作業(yè)的最大并行度，如果資源不夠，及時擴容。?作業(yè)運行時間：根據(jù)作業(yè)的運行時間來判斷作業(yè)是否存活，中途是否掉線過。?Checkpoint 情況：Checkpoint 是 JobManager 發(fā)起的，并且關(guān)乎到作業(yè)的狀態(tài)是否可以完整的保存。
TaskManager
因為所有的作業(yè)最終都是運行在 TaskManager 上，所以 TaskManager 的監(jiān)控指標也是異常的監(jiān)控，并且作業(yè)的復(fù)雜度也會影響 TaskManager 的資源使用情況，所以 TaskManager 的基礎(chǔ)監(jiān)控指標比如內(nèi)存、GC 如果出現(xiàn)異常或者超出設(shè)置的閾值則需要立馬進行告警通知，防止后面導(dǎo)致大批量的作業(yè)出現(xiàn)故障重啟。
?內(nèi)存使用率：部分作業(yè)的算子會將所有的 State 數(shù)據(jù)存儲在內(nèi)存中，這樣就會導(dǎo)致 TaskManager 的內(nèi)存使用率會上升，還有就是可以根據(jù)該指標看作業(yè)的利用率，從而最后來重新劃分資源的配置。?GC 情況：分時間和次數(shù)，一旦 TaskManager 的內(nèi)存率很高的時候，必定伴隨著頻繁的 GC，如果在 GC 的時候沒有得到及時的預(yù)警，那么將面臨 OOM 風險。
Flink Job
作業(yè)的穩(wěn)定性和及時性其實就是大家最關(guān)心的，常見的指標有：作業(yè)的狀態(tài)、Task 的狀態(tài)、作業(yè)算子的消費速度、作業(yè)出現(xiàn)的異常日志。
?作業(yè)的狀態(tài)：在 UI 上是可以看到作業(yè)的狀態(tài)信息，常見的狀態(tài)變更信息如下圖所示。
?Task 的狀態(tài)：其實導(dǎo)致作業(yè)的狀態(tài)發(fā)生變化的原因通常是由于 Task 的運行狀態(tài)出現(xiàn)導(dǎo)致，所以也需要對 Task 的運行狀態(tài)進行監(jiān)控，Task 的運行狀態(tài)如下圖所示。
?作業(yè)異常日志：導(dǎo)致 Task 出現(xiàn)狀態(tài)異常的根因通常是作業(yè)中的代碼出現(xiàn)各種各樣的異常日志，最后可能還會導(dǎo)致作業(yè)無限重啟，所以作業(yè)的異常日志也是需要及時關(guān)注。?作業(yè)重啟次數(shù)：當 Task 狀態(tài)和作業(yè)的狀態(tài)發(fā)生變化的時候，如果作業(yè)中配置了重啟策略或者開啟了 Checkpoint 則會進行作業(yè)重啟的，重啟作業(yè)的帶來的影響也會很多，并且會伴隨著一些不確定的因素，最終導(dǎo)致作業(yè)一直重啟，這樣既不能解決問題，還一直在占用著資源的消耗。?算子的消費速度：代表了作業(yè)的消費能力，還可以知道作業(yè)是否發(fā)生延遲，可以包含算子接收的數(shù)據(jù)量和發(fā)出去數(shù)據(jù)量，從而可以知道在算子處是否有發(fā)生數(shù)據(jù)的丟失。
小結(jié)與反思
本節(jié)講了 Flink 中常見的監(jiān)控對象，比如 JobManager、TaskManager 和 Flink Job，對于這幾個分別介紹了其內(nèi)部大概有的監(jiān)控指標，以及在真實生產(chǎn)環(huán)境關(guān)心的指標，你是否還有其他的監(jiān)控指標需要補充呢？
本節(jié)涉及的監(jiān)控指標對應(yīng)的含義可以參考官網(wǎng)鏈接：metrics
本節(jié)涉及的監(jiān)控指標列表地址：https://github.com/zhisheng17/flink-learning/blob/master/flink-learning-monitor/flink_monitor_measurements.md
--end--