鑽石舞台

流計算作業通常運行時間長，數據吞吐量大，且對時延較為敏感。但實際運行中，Flink作業可能因為各種原因出現吞吐量抖動、延遲高、快照失敗等突發情況，甚至發生崩潰和重啟，影響輸出數據的質量，甚至會導致線上業務中斷，造成報表斷崖、監控斷點、數據錯亂等嚴重後果。

本文會對Flink常見的問題進行現象展示，從原理上說明成因和解決方案，並給出線上問題排查的工具技巧，幫助大家更好地應對Flink 作業的異常場景。

如何分析 Flink問題？

下圖描述了遇到 Flink問題時，建議的處理步驟：

發生問題時，首先要做的是現象記錄，即檢查作業的運行狀態。如果運行狀態不是運行中，那肯定沒有數據正常輸出了，需要進一步從日誌中查找問題根因。如果作業在運行中，但是存在近期的重啟記錄，也表明可能發生了較嚴重的問題。此時需要整理問題發生的時間線，便於後續定位參考。

作業的吞吐和延時等指標是作業運行是否正常的判斷標準。如果一個運行中的作業輸出中斷、數據量變小等現象，則首先需要觀察是否存在嚴重的背壓（也稱反壓，即 BackPressure. 後文會細講如何判定）。如果存在背壓，則需根據定位表，找到問題算子並進行瓶頸分析定位。隨後還可以查看快照的時長和大小等信息，如果快照過大（例如大於 1GB）或很長時間才完成，則可能對內存造成較大壓力。

如果從指標上不能完全判斷問題原因，則需要結合完整的日誌進行更細緻的追查。後文會提到定位異常時常見的報錯關鍵字，可以提升問題定位的速度。如果日誌中沒有太多有用的信息，則還需要對作業運行的環境進行檢查，例如排除是否有其他進程干擾，系統是否被重啟過，網絡和磁盤是否存在瓶頸等等…

這裡我們總結了Flink作業的常見故障、確認方法和建議的解決措施。圖中的 JM 表示 JobManager，TM 表示 TaskManager。

1.作業自動停止

現象：本應長期運行的作業，突然停止運行，且再也不恢復。

如果 Flink 作業在編程時，源算子實現不當，則可能造成源算子處理完數據以後進入 FINISHED 狀態。如果所有源算子都進入了 FINISHED 狀態，那整個 Flink 作業也會跟着結束。

Flink 作業默認的容錯次數是 2，即發生兩次崩潰後，作業就自動退出了，不再進行重試。當出現此種場景時，TaskManager 的日誌中會有「restart strategy prevented it」字樣。我們首先要找到作業崩潰的原因，其次可以適當調大 RestartStrategy 中容錯的最大次數，畢竟節點異常等外部風險始終存在，作業不會在理想的環境中運行。

此外，舊版Flink（低於 1.11.0）的 RocksDB 內存使用不受管控，造成很容易由於超量使用而被外界（YARN、Kubernetes 等）KILL 掉。如果經常受此困擾，可以考慮升級 Flink 版本到最新，其默認開啟自動內存管理功能。

2.輸出量穩定但不及預期

現象：作業輸出量較穩定，但是不及預期值（正常情況下，每核 5000 ~ 20000 條/秒）。

如果作業輸出量達不到預期，我們需要分別從 CPU、內存、磁盤、網絡等方面逐一排查是否遇到了瓶頸。

CPU 的瓶頸通常是因為序列化、反序列化開銷較大，或者用戶自定義算子的某個方法的時間複雜度高。CPU 瓶頸的定位較為簡單，使用 JProfiler、jvmtop.sh 等工具均能較為準確的找到原因。

如果發現內存占比過高，那通常伴隨着較長的 GC 時間，或者較多的 FullGC 次數。內存分析可以通過 jmap把堆內存 dump下來，然後使用 MemoryAnalyzer等自動化工具進行泄露分析。

如果是因為數據傾斜原因，則網上已經有不少通用的解決方法，例如 key打散、預聚合（Flink中叫做 Local-GlobalAggregation）等，建議結合具體業務場景選用。

還有一個常見的拖累吞吐的原因是訪問外部（第三方）系統。假設每條數據都需要訪問外部系統，每次需要 1ms，那麼 1s只能處理 1000 條數據，這當然是非常少的。如果需要頻繁訪問外部系統的話，建議充分利用批量存取和緩存、異步算子等功能，儘可能地減少交互次數。

3.輸出量逐步減少或完全無輸出

現象：作業輸出量一開始較高，後來越來越少，甚至降到 0.

作業輸出量逐步減少的原因，最常見是背壓較高和 Full GC 時間太長。當一個算子遇到 CPU 或者 I/O 瓶頸時，會造成輸入緩衝區的數據積壓，這樣它的上游（運行圖中的前一個算子）的輸出緩衝區也會發生積壓。就這樣，一級一級向前傳遞，就會導致從數據源到問題算子的一條鏈路的數據都發生積壓，這就是出現了「背壓」現象。當然，如果算子的輸出緩衝區寫不出去（網絡質量太差），也是可能引發背壓的。

當我們在 Flink Web UI 界面上發現背壓後，我們可以用後文中的「背壓分析表」來定位可能的問題節點。

另外，如果沒有發現嚴重的背壓，但是數據輸出量還是很少，就需要檢查數據源、數據目的本身是否有問題。例如我們曾遇到過 MySQL 連接數滿了導致數據源無法消費，或者下游數據目的經常連接超時造成數據無法穩定輸出等。這些問題的排查思路都是控制變量法，去掉其他算子，構造一個單純用到Source或Sink的作業，然後觀察問題是否仍然存在。

最後，如果在日誌里看到有數據錯誤的報錯，尤其是那種瘋狂寫日誌的場景，請務必引起重視。異常數據（數據輸入格式與定義的 Schema 不一致）會造成計算結果錯亂，還會造成磁盤空間被異常日誌占滿等嚴重問題。

如果我們發現算子背壓較高，而且內存用量很大，JVM Full GC 時間很長，則說明堆內存用量太大了。Flink 的堆內存除了框架層面使用外，主要是用戶定義的狀態（含窗口等間接用到的狀態）和運行時臨時創建的對象占用了大部分內存。

當狀態過多時，如果啟用了快照（Checkpoint），就會發現每次快照完成後的狀態都很大，而且所需時間也較長。Flink 在快照過程中，會對所有狀態做全量讀取，如果是異步快照的話還有 Copy-On-Write 操作帶來的內存壓力，因此如果快照過大或者用時較長，也會造成內存中大量對象長期停留而無法被 GC 清理。

為了加速快照的執行，可以啟用增量快照（目前只有 RocksDB State Backend 支持），或者如果有自定義快照的邏輯，請儘量避免 snapshotState() 方法耗時過長。另外如果在使用最新版本的 Flink（1.11 及以上），則可以開啟 Unaligned Checkpoint 特性，該特性可以避免多個輸入流的速度不同時（例如 JOIN 操作）快照帶來的停頓和數據暫存開銷。

窗口、GROUP BY 等算子（語句）都會用到大量狀態數據，因此如果定義窗口的話，建議不要設置太大的窗口，或者太小的滑動時間（僅針對 Sliding Window 而言）。如果因為業務邏輯原因不得不用，則需要設置 Idle State Retention Time 以定期清理失效的狀態。

如果用到了自定義的狀態對象（StateDescriptor），則一定不要忘記清理或者設置 State TTL 以令Flink 自動清理過期的狀態。

還有一個不太常見但是會造成輸出數據突然減少的原因是 Watermark 錯亂。通常情況下我們的 Watermark 是基於輸入數據時間戳來計算的，如果輸入數據有明顯的異常時間戳（例如 2050 年的某一天），則會將 Watermark 直接快進到那一天，從而令後續的正常數據被當作過期數據丟掉了。這就需要我們妥善定義 Watermark 的生成策略（忽略或矯正異常時間戳），或者對數據源的時間戳字段先做一遍清洗校驗。相反，如果輸入數據的時間戳一直不變（常見於測試數據，一直輸入同一條），則會造成 Watermark 長期無法超過窗口的邊界，這樣窗口也會久久無法觸發計算，從外部來看就是沒有數據輸出。

4.個別數據缺失

現象：作業輸出整體穩定，但是個別數據缺失，造成結果的精度下降，甚至結果完全錯亂。

當遇到懷疑數據缺失造成的計算結果不正確時，首先需要檢查作業邏輯是否不小心過濾了一些正常數據。檢查方法可以在本地運行一個 Mini Cluster，也可以在遠端的調試環境進行遠程調試或者採樣等。具體技巧後文也會提到。

另外還有一種情況是，如果用戶定義了批量存取的算子（通常用於與外部系統進行交互），則有可能出現一批數據中有一條異常數據，導致整批次都失敗而被丟棄的情況。

對於數據源 Source 和數據目的Sink，請務必保證 Flink 作業運行期間不要對其進行任何改動（例如新增 Kafka 分區、調整 MySQL 表結構等），否則可能造成正在運行的作業無法感知新增的分區或者讀寫失敗。儘管 Flink 可以開啟 Kafka 分區自動發現機制（在 Configuration 里設置 flink.partition-discovery.interval-millis值），但分區發現仍然需要一定時間，數據的精度可能會稍有影響。

5.作業頻繁重啟

現象：作業頻繁重啟又自行恢復，陷入無盡循環，無法正常處理數據。

作業頻繁重啟的成因非常多，例如異常數據造成的作業崩潰，可以在 TaskManager 的日誌中找到報錯。數據源或者數據目的等上下游系統超時也會造成作業無法啟動而一直在重啟。此外 TaskManager Full GC 太久造成心跳包超時而被 JobManager 踢掉也是常見的作業重啟原因。如果系統內存嚴重匱乏，那麼 Linux 自帶的 OOM Killer 也可能把 TaskManager 所在的 JVM 進程 kill 了。

當一個正常運行的作業失敗時，日誌里會有 from RUNNING to FAILED 的關鍵字，我們以此為着手點，查看它後面的 Exception 原因，通常最下面的 caused by 即是直接原因。當然，直接原因不一定等於根本原因，後者需要藉助下文提到的多項技術進行分析。

如果 JVM 的內存容量超出了平台方（例如 YARN 或 Kubernetes 等）的容器限制，則可能被 KILL。問題的確認方式也是查看作業日誌以及平台組件的運行狀態。值得一提的是，在最新的 Flink 版本中，只要設置 taskmanager.memory.process.size 參數，基本可以保證內存用量不會超過該值（前提是用戶沒有使用 JNI 等方式申請 native 內存）。

作業的崩潰重啟還有一些原因，例如使用了不成熟的第三方 so 庫，或者連接數過多等，都可以從日誌中找到端倪。

上面小節總結了 Flink 作業異常的常見現象和可能的原因，下面我們來介紹一下定位問題時常用的小工具和技巧，這對分析性能瓶頸非常有用。

1.常用工具內存

•堆內：jcmd、jmap、jstat、MemoryAnalyzer

•堆外（Direct）：Natve Memory Tracking(NMT)

•Native：jemalloc（jeprof）、tcmalloc（pprof）

對於堆內內存，我們可以用 jcmd命令開啟 JavaFlightRecorder (JFR) 的錄製功能，它會把 JVM 運行期間的各項指標等都保存在文件中，類似飛機的「黑匣子」，可以後續分析。jmap命令則可以把堆內存 dump出來，隨後可以配合 MemoryAnalyzer分析是否有內存泄漏、占內存過多的對象等。jstat命令則可以打印 GC 的統計指標，便於我們觀察 GC 是否正常。

對於 JVM 的堆外內存，通常由 netty等使用的 DirectByteBuffer占用，可以使用 JVM 自帶的 Native Memory Tracking功能來記錄和打印這些內存對象的分配情況。不過正常情況下用戶代碼不會涉及到這部分的內存。

如果使用 RocksDB 或者 JNI 調用了第三方的 so庫，那有可能會用到 malloc函數。這樣分配的內存是不受 JVM 管控的，因此如果需要定位這裡的問題，需要使用 jemalloc或 tcmalloc動態替換原生的 malloc 實現，並開啟 profiling以追蹤內存分配。

CPU

•工具： JVisualVM、JProfiler、jstack、Arthas、JFR、jvmtop

很多工具都可以查看運行時 CPU 的使用情況。如果我們觀察到 JVM 所在進程的 CPU 很繁忙，則需要找出熱點方法（最高頻、最耗時的）。如果空閒較多，則需要分析是否出現了死鎖、I/O 等待等問題。

特別值得一提的是，jvmtop是一個很好用的小工具，可以查看哪些方法占用 CPU 最高，並加以排序，這樣我們可以很直觀的找出潛在的問題點。JProfiler、Arthas則是大而全的工具箱，裡面提供了非常多的實用功能。

圖：使用 JProfiler 分析熱點方法

磁盤 I/O

•iotop、dstat

如果想了解磁盤的使用情況，則可以用 iotop等工具來查看 JVM 進程的磁盤讀取和寫入量。dstat命令則可以持續的輸出系統整體的磁盤讀寫情況。

網絡 I/O

•查看網絡資源占用情況：nload、nethogs、iftop

•檢查是否丟包：tcpdump 抓包查看（注意內核緩衝區大小）

•定位連接數過多的問題：netstat、ss

2.指標分析

Flink 指標通常可以在自帶的 Web UI 中查看，也可自定義 Metric Reporter，將指標輸出到第三方系統，例如 Prometheus、InfluxDB、Elasticsearch等等，隨後可以展示為報表或進行告警等。

重點關注的算子指標

•源算子（Source）：numRecordsOutPerSec（每秒產生的數據條數）、 numRecordsOut（產生的數據總條數）

•目的算子（Sink）：numRecordsInPerSec（每秒接收的數據條數）、numRecordsIn（接收的數據總條數）

•其他算子：除了觀察上述吞吐量指標外，還可以觀察 Watermark（時間戳是否符合預期）、Backpressure（紅色 HIGH 表示背壓高）。

圖：查看算子的 Watermark、Backpressure和各項統計指標

背壓分析

首先我們來看一下為什麼會出現背壓高的現象。Flink 的每個算子都有輸入緩衝區（InPool）和輸出緩衝區（OutPool），它們的使用率分別在 Flink 指標里叫做 inPoolUsage和 outPoolUsage。

特別提一下，在 Flink 1.9 及更高版本，inPoolUsage還細分為 exclusiveBufferUsage（每個Channel獨占的 Buffer）和 floatingBufferUsage（按照 Channel需求，動態分配和歸還的 Buffer）。

因此，我們可以用下面的兩個表格來定位背壓產生的位置和可能原因（圖片來自Flink Network Stack Vol. 2: Monitoring, Metrics, and that Backpressure Thing一文）。

如果 inPoolUsage較低，而 outPoolUsage也較低，則說明完全沒有背壓現象。若 inPoolUsage較低，而 outPoolUsage卻很高，則說明處於臨時狀態，可能是背壓剛開始，也可能是剛結束，需要再觀察。

若 inPoolUsage較高，而 outPoolUsage低，那麼通常情況下這個算子就是背壓的根源了。但如果 inPoolUsage較高，而 outPoolUsage也較高的話，則說明這個算子是被其他下游算子反壓而來的，並不是元兇。

而對於 Flink 1.9 等以上版本，我們還可以用 floatingBufferUsage 和 exclusiveBufferUsage來進一步定位問題，方法和上面的一致，即：首先看 floatingBufferUsage，然後看上游的 outPoolUsage，隨後再看 exclusiveBufferUsage，這樣就可以按表格中的對應項找到解答了：

當我們找到背壓的算子以後，還需要用到上文提到的常用工具，進一步的定位產生的根源。

特別要注意的是，在背壓定位過程中，建議關閉 Operator Chaining優化，這樣所有的算子可以單獨拆分出來，不至於相互干擾。

3.日誌分析收集哪些日誌

•JobManager 日誌

•TaskManager 日誌

•GC 日誌（-XX:+PrintGCDetails-XX:+PrintGCDateStamps）

•YARN / Kubernetes 日誌

•系統日誌（/var/log 下的日誌，以及 journalctl 等）

問題定位關鍵字

•from RUNNING to FAILED 可以查看作業崩潰的直接原因

•exit 可以查看進程的 ExitCode

•fatal 或 Fatal 可以看 JVM Core Dump 的報錯，或者 Akka 報錯

•shutting down JVM 可以看 Akka 的 akka.jvm-exit-on-fatal-error 報錯

•java.lang.OutOfMemoryError 可以查看是否發生過 OOM

•timeout 或 Timeout 表示發生了超時，此時可以查看網絡質量，確認是否存在大量丟包等情況

•Failure 可以查看 Checkpoint 失敗的信息

•搜索 Total time for which application threads were stopped: ，查看後面是否有很長時間的停頓。這個是 Full GC 導致的全局停止時間

•Kill 可以看到超用資源而被 YARN 強行 kill 的情況

•Exception 則可以看到其他的異常（不一定是原因，僅供參考）

•WARN ERROR 等則可以看到一些報錯的日誌（也不一定是原因）

日誌中常見的錯誤碼

•239（-17）：Fatal uncaught exception, 通常是 OOM。

•1：組件初始化時出錯。

•2：組件運行時報錯，通常有 Fatal error occurred in the cluster entrypoint 字樣。

•1443：作業狀態變成 FAILED 時會出現，可搜索 to FAILED 尋找原因。

•243（-13）：嚴重錯誤，較少見，通常有 FATAL ERROR 字樣。

•31：命令行解析錯誤，或者 YARN 初始化錯誤，通常不會遇到。

•128~159 通常是 KILL 信號導致的。例如 134 (core dumped SIGABRT) 可能是 JVM 異常，也可能是第三方 so 的問題。

•特別地，137 是 SIGKILL（kill -9 導致，可能是 OOM Killer 或者人工調用），143 是普通 SIGTERM，可能是 YARN Kill，也可能是人工調用。

本文講述了 Flink 問題定位的思路，以及常見的問題現象和解決方案，以及一些定位的小技巧。希望能夠通過本文，增強大家對 Flink 的常見問題定位的思路。

此外，如果遇到了難以解決的問題，通過上述的分析還是解決不了的話，還可以通過向社區發郵件（https://flink.apache.org/zh/community.html）的方式來獲取幫助。社區的 Committer通常會很快回復，如果確認是 Flink 的 bug的話，則可以提交一個 JIRA 單來追蹤這個問題。需要注意的是，提問時應當準確描述問題的現象、Flink 版本、最小復現方式等，最好可以附上日誌和運行的環境等信息。

最後，祝各位 Flink玩的愉快 ：）