這篇部落格文章介紹了Apache Flink的内置監控和名額系統，使開發人員能夠有效地監控他們的Flink程式。通常情況下，對于剛開始使用流處理和Apache Flink的DevOps團隊，選擇相關名額來監控Flink程式可能是比較困難的。在參與部署過許多大規模Flink環境後，我想在這裡與社群分享我的經驗和一些最佳實踐。
對于在Apache Flink上運作的關鍵業務程式，其性能監控成為生産環境中越來越重要的部分。它可確定快速發現并處理業務程式的性能下降、異常停止等問題。
監控與可觀察性密切相關，可觀察性是故障排除和性能調優的先決條件。如今，随着現代企業應用程式的複雜性和傳遞速度的提高，工程團隊必須在任何給定的時間點了解并全面了解其應用程式的狀态。
1.Flink的度量系統
監控Flink程式的基礎是其
度量系統，它由兩部分組成：度量名額和度量報告。
1.1 度量名額
Flink提供了一整套内置名額，例如：
- 使用的JVM堆/非堆/直接記憶體（每個TaskManager/JobManager）
- 程式重新開機次數（每個程式）
- 每秒記錄數（每個算子）
- ...
這些名額具有不同的衡量範圍，不僅包含Flink本身，還有更普遍的JVM，作業系統名額。
作為使用者，可以為自己的功能添加特定應用程式的名額。通常這些包括無效記錄數計數器或在State中臨時緩沖的記錄數計數器。除了計數器，Flink還提供其他名額類型，如儀表和直方圖。有關如何使用Flink的名額系統注冊自己的名額的說明，請檢視
Flink的文檔。在這篇部落格文章中，我們将重點介紹如何充分利用Flink的内置名額。
1.2 度量報告
通過Flink的REST API可以查詢所有名額，同時使用者可以配置度量報告系統，将名額發送到外部系統。Apache Flink為度量報告提供了開箱即用的最常見的監控工具，包括JMX，Prometheus，Datadog，Graphite和InfluxDB。有關如何配置報告的資訊，請檢視Flink的MetricsReporter文檔。
在本博文的其餘部分，我們将介紹一些最重要的名額來監控你的Apache Flink應用程式。
2.監測名額
要監控的第一件事是你的程式是否實際處于運作狀态。此外，它還可以監控重新開機次數和自上一次重新開機的時間。
一般來說，成功的檢查點是檢測應用程式整體狀況的關鍵因素。對于每個檢查點，檢查點擋闆需要流經Flink程式的整個拓撲，拓撲中正常的業務事件和擋闆按順序處理不能互相超越。是以，成功的檢查點可以表示沒有通道被完全阻塞。
關鍵名額
| 度量名額 | 範圍 | 描述 |
|---|---|---|
| uptime | 程式 | 程式沒有中斷的運作時長。 |
| fullRestarts | 自送出此程式以來完全重新啟動的總次數。 | |
| numberOfCompletedCheckpoints | 成功完成檢查點的數量。 | |
| numberOfFailedCheckpoints | 失敗檢查點的數量。 |
示例儀表闆面闆
正常運作時間（35分鐘），重新開機時間（3毫秒）和完全重新開機次數（7）
完成檢查點（18336），失敗（14）
可能的警報
- ΔfullRestarts > threshold
- ΔnumberOfFailedCheckpoints > threshold
3.監控進度和吞吐量
現在可以知道你的程式是否正在運作，它的檢查點是否正常完成，但還不能知道程式是否在不斷消費，能否跟得上上遊系統資料的産生速度。
3.1 吞吐量
Flink提供多個名額來衡量我們的應用程式的吞吐量。一個程式可以包含多個鍊式任務 Flink計算進出的記錄和位元組數。在這些度量中，每個操作員的傳出記錄的速率通常是最直覺和最容易推理的。對于每一個算子或任務（一個任務可以包含多個串起來的子任務），Flink記錄了進入和寫出的記錄數和位元組數大小。在這些度量名額中，每個算子的寫出速率通常是最直覺，最容易了解的。
| numRecordsOutPerSecond | 任務 | 此任務每秒發送的記錄數。 |
| 算子 | 此算子每秒發送的記錄數。 |
每個算子每秒平均記錄數
- recordsOutPerSecond= 0（對于非Sink算子）
注意：Source算子始終沒有傳入記錄，Sink算子始終沒有傳出記錄，因為度量标準僅計算Flink内部通信。不過
JIRA Ticket可以改變這個情況。
3.2 進展
對于使用事件時間語義的應用程式，随時間推移水位就顯得非常重要。時間水位t告訴整個程式，它不應再期望接收時間戳早于t的事件，并且觸發程式排程時間戳< t的所有操作 。例如，一旦水位通過30，程式将關閉并計算在t = 30 結束的時間視窗内的事件。
是以，在應用程式中對事件時間敏感的算子需要監控水位，例如過程函數和視窗。如果目前處理時間和水位之間的差異非常高，那麼它通常意味着兩個問題。第一，它可能意味着你正在處理舊事件，例如在停機後追資料或程式處理速度沒辦法趕上上遊資料産生速度，上遊事件堆積。第二，這可能意味着單個上遊子任務長時間沒有發送水位（例如，因為它沒有接收任何基于水位的事件），這也阻止了下遊算子的水位處理。這個
為後者提供了進一步的資訊和解決方案。
| currentOutputWatermark | 此算子發出的最後一個水印。 |
拓撲中單個算子的每個子任務的事件時間延遲。在這種情況下，對于每個子任務，時間水位落後幾秒鐘。
- currentProcessingTime - currentOutputWatermark > threshold
3.3 “緊跟”
當從消息隊列中進行消費時，通常可以直接監視應用程式是否跟得上消息産生速度。通過使用特定連接配接器的度量名額，你可以監視目前消費者組的消息與最新消息的差距。Flink可以基于大多數Source連接配接器轉發其基礎名額。
| records-lag-max | 使用者 | 适用于FlinkKafkaConsumer。此視窗中任何分區的記錄數最大延遲。随着時間的推移，越來越大的延時表明消費者沒有跟上生産者的速度。 |
| millisBehindLatest | 适用于FlinkKinesisConsumer。消費者距離最新消息的毫秒數。對于任何消費者和Kinesis分片，這表示它與目前時間之間的差距。 |
- records-lag-max > threshold
- millisBehindLatest > threshold
4.監控時延
一般而言，時延是指事件建立與基于此事件的結果變得可見之間的時間延遲。建立事件後，它通常存儲在持久性消息隊列中，然後由Apache Flink處理，将結果寫入資料庫或調用下遊系統。在這樣的資料處理管道中，可以在每個階段引入時延。原因有多種，包括：
- 在事件持久存儲在消息隊列中之前，可能需要不同的時間。
- 在高負載期間或恢複期間，事件可能會在消息隊列中花費一些時間，直到Flink處理它們（請參閱上一節）。
- 出于功能原因，流式拓撲中的一些函數需要緩沖事件一段時間（例如，在時間視窗中）。
- Flink拓撲（架構或使用者代碼）中的每個計算以及每個網絡shuffle都需要時間并增加時延。
- 如果應用程式通過事務型Sink節點寫出，則Sink節點将僅在Flink的成功檢查點上送出和下發事務，每個檢查點之間的間隔時間也将增加時延。
實際上，事實證明，在多個階段（事件建立，存儲，進入Flink，寫出Flink，若資料量過大可以隻采樣部分資料）為事件添加時間戳是非常有價值的。這些時間戳之間的差異可以作為Flink拓撲中的使用者自定義度量标準展示，以獲得每個階段的延遲分布情況。
在本節的剩下部分，我們隻考慮在Flink拓撲中的時延，但并不包括事務型sink節點或由于函數原因緩存事件的節點。
為此，Flink提供了一項稱為
時延跟蹤的功能。啟用後，Flink将在所有來源定期插入延遲标記事件。對于每個子任務，将報告從每個源到此算子的延遲分布。可以通過根據需要設定metrics.latency.granularity來進一步控制這些直方圖的粒度。
由于可能存在大量的直方圖（特别是對于 metrics.latency.granularity：子任務），啟用延遲跟蹤會顯着影響群集的性能。建議僅在調試期間使其能夠找到延遲源。
度量
| latency | 從源算子到此算子的時延。 | |
| restartingTime | 重新啟動程式所花費的時間，或目前重新啟動的持續時間。 |
Source和單個Sink子任務之間的時延分布。
4.JVM名額
到目前為止，我們隻關注了Flink特定的名額。隻要你的應用程式時延和吞吐量符合你的期望并且檢查點持續正常，整個程式應該是沒有問題的。但另一方面，如果程式性能開始下降，你首要考慮的名額就是TaskManager＆JobManager JVM的記憶體消耗和CPU負載。
4.1 記憶體
Flink報告了JobManagers和TaskManagers的Heap，NonHeap，Direct和Mapped記憶體的使用情況。
- 堆記憶體 - 與大多數JVM應用程式一樣 - 是最易于觀察的重要名額。尤其是在使用Flink的檔案系統statebackend時，因為它将所有狀态對象保留在JVM堆上。如果堆上的對象大小顯着增加，這通常可歸因于應用程式狀态的大小（檢查堆棧狀态的估計大小的 檢查點名額 ）。增長狀态的可能原因是特定于應用程式的。通常，越來越多的主鍵，不同輸入流之間的事件時間偏差過大或者僅僅缺少狀态清理都可能導緻整個堆呈增長狀态。
- NonHeap記憶體由元空間控制，預設情況下其大小不受限制，并儲存類中繼資料和靜态内容。有一個 預設将大小限制為250兆位元組。
- 直接記憶體的最大驅動因素是Flink的網絡緩沖區數量，可以 配置 。
- 映射記憶體通常接近零，因為Flink不使用記憶體映射檔案。
在容器化環境中，你還應監視JobManger和TaskManager容器的總體記憶體消耗，以確定它們不超過其資源限制。當使用RocksDB狀态後端時，要注意RocksDB會從堆下配置設定大量記憶體。如果想了解更多關于RocksDB使用記憶體量的多少，可以檢視 Stefan Richter 撰寫的
這篇部落格文章| Status.JVM.Memory.NonHeap.Committed | 程式/任務管理器 | 保證JVM可用的非堆記憶體量（以位元組為機關）。 |
| Status.JVM.Memory.Heap.Used | 目前使用的堆記憶體量（以位元組為機關）。 | |
| Status.JVM.Memory.Heap.Committed | Job-/TaskManager | 保證可供JVM使用的堆記憶體量（以位元組為機關）。 |
| Status.JVM.Memory.Direct.MemoryUsed | JVM用于直接緩沖池的記憶體量（以位元組為機關）。 | |
| Status.JVM.Memory.Mapped.MemoryUsed | 執行G1 Young Generation垃圾收集所花費的總時間。 | |
| Status.JVM.GarbageCollector.G1 Young Generation.Time | ||
| Status.JVM.GarbageCollector.G1 Old Generation.Time | 執行G1 Old Generation垃圾收集所花費的總時間。 |
TaskManager記憶體消耗和垃圾收集時間。
JobManager記憶體消耗和垃圾收集時間。
- container memory limit < container memory + safety margin
4.2 CPU
除了記憶體，你還應該監視TaskManagers的CPU負載。如果你的TaskManagers經常處于非常高的負載下，你可以通過減少每個TaskManager的任務槽數來提高整體性能（如果是Standalone部署），為TaskManager提供更多資源（如果是容器化部署），或提供更多的TaskManagers。通常，在正常業務期間已經在非常高的負載狀态下運作的系統，在從停機時間恢複之後将需要更多的時間來追資料。在此期間，你将看到比平時更高的延遲（事件時間偏差）。
CPU負載的突然增加也可能是因為過多垃圾收集壓力，這在JVM記憶體名額中可見。
如果一個或幾個TaskManagers一直處于非常高的負載下，由于長檢查點對齊時間和事件時間偏差增加，這可能會降低整個拓撲結構的速度。常見的原因是資料分區鍵的資料傾斜，這可以通過在shuffle之前預先聚合或者将分區鍵更改為均勻分布的主鍵上來減輕資料傾斜。
| 公 | ||
|---|---|---|
| Status.JVM.CPU.Load | JVM最近的CPU使用情況。 |
TaskManager和JobManager CPU加載。
5.系統資源
除了上面的JVM名額之外，還可以使用Flink的名額系統來收集有關系統資源，即整個計算機的記憶體，CPU和網絡相關名額，而不僅僅是Flink程式。預設情況下禁用系統資源監視，并且需要對類路徑有其他依賴性。可以檢視
Flink系統資源名額文檔以擷取相關指導和詳細資訊。Flink中的系統資源監視在沒有現有主機監視功能的設定中非常有用。
6.結論
這篇文章主要闡明Flink的名額和監控系統。當你是第一次考慮如何成功實作監控Flink應用程式時，可以通過這方面的知識學習作為起點。我更推薦在開發階段的早期就開始監控你的Flink應用程式。這樣能夠随着時間的推移改進儀表闆和警報，更重要的是，你在整個開發階段觀察應用程式更改對性能的影響。通過這樣做你可以提出有關應用程式運作時行為的正确問題，在早期就可以了解到有關Flink内部的更多資訊。
最後想說的是，這篇文章僅涉及Apache Flink的整體名額和監控功能。本人建議可以浏覽
Flink的名額文檔，以擷取Flink名額系統的完整參考。
本文由阿裡雲開發者社群組織翻譯。
文章原标題《Monitoring Apache Flink Applications 101》
作者：AJ Christensen
譯者：麼凹
校對：校對者：楊陽（時溪）
文章為簡譯，更為詳細的内容，請檢視
原文