今天就跟大家聊聊有關如何進行Flink作業問題分析和調優實踐,可能很多人都不太了解,為了讓大家更加了解,小編給大家總結了以下內容,希望大家根據這篇文章可以有所收獲。
創新互聯主要從事成都做網站、成都網站制作、網頁設計、企業做網站、公司建網站等業務。立足成都服務秦皇島,10多年網站建設經驗,價格優惠、服務專業,歡迎來電咨詢建站服務:13518219792
1.什么是 checkpoint
簡單地說就是 Flink 為了達到容錯和 exactly-once 語義的功能,定期把 state 持久化下來,而這一持久化的過程就叫做 checkpoint ,它是 Flink Job 在某一時刻全局狀態的快照。當我們要對分布式系統實現一個全局狀態保留的功能時,傳統方案會引入一個統一時鐘,通過分布式系統中的 master 節點廣播出去給每一個 slaves 節點,當節點接收到這個統一時鐘時,它們就記錄下自己當前的狀態即可。但是統一時鐘的方式也存在一定的問題,某一個 node 進行的 GC 時間比較長,或者 master 與 slaves 的網絡在當時存在波動而造成時鐘的發送延遲或者發送失敗,都會造成此 slave 和其它的機器出現數據不一致而最終導致腦裂的情況。如果我們想要解決這個問題,就需要對 master 和 slaves 做一個 HA(High Availability)。但是,一個系統越是復雜,就越不穩定且維護成本越高。Flink 是將 checkpoint 都放進了一個名為 Barrier 的流。上圖中就是一個 Barrier 的例子,從上游的第一個 Task 到下游的最后一個 Task,每次當 Task 經過圖中藍色的柵欄時,就會觸發 save snapshot(快照)的功能。我們用一個例子來簡單說明。2.實例分析
這是一個簡單的 ETL 過程,首先我們把數據從 Kafka 中拿過來進行一個 trans 的轉換操作,然后再發送到一個下游的 Kafka此時這個例子中沒有進行 chaining 的調優。所以此時采用的是 forward strategy ,也就是 “一個 task 的輸出只發送給一個 task 作為輸入”,這樣的方式,這樣做也有一個好處就是如果兩個 task 都在一個 JVM 中的話,那么就可以避免不必要的網絡開銷設置 Parallism 為 2,此時的 DAG 圖如下:■ CK的分析過程
每一個 Flink 作業都會有一個 JobManager ,JobManager 里面又會有一個 checkpoint coordinator 來管理整個 checkpoint 的過程,我們可以設置一個時間間隔讓 checkpoint coordinator 將一個 checkpoint 的事件發送給每一個 Container 中的 source task,也就是第一個任務(對應并行圖中的 task1,task2)。當某個 Source 算子收到一個 Barrier 時,它會暫停自身的數據處理,然后將自己的當前 state 制作成 snapshot(快照),并保存到指定的持久化存儲中,最后向 CheckpointCoordinator 異步發送一個 ack(Acknowledge character --- 確認字符),同時向自身所有下游算子廣播該 Barrier 后恢復自身的數據處理。每個算子按照上面不斷制作 snapshot 并向下游廣播,直到最后 Barrier 傳遞到 sink 算子,此時快照便制作完成。這時候需要注意的是,上游算子可能是多個數據源,對應多個 Barrier 需要全部到齊才一次性觸發 checkpoint ,所以在遇到 checkpoint 時間較長的情況時,有可能是因為數據對齊需要耗費的時間比較長所造成的。■ Snapshot & Recover
如圖,這是我們的Container容器初始化的階段,e1 和 e2 是剛從 Kafka 消費過來的數據,與此同時,CheckpointCoordinator 也往它發送了 Barrier。此時 Task1 完成了它的 checkpoint 過程,效果就是記錄下 offset 為2(e1,e2),然后把 Barrier 往下游的算子廣播,Task3 的輸入為 Task1 的輸出,現在假設我的這個程序的功能是統計數據的條數,此時 Task3 的 checkpoint 效果就是就記錄數據數為2(因為從 Task1 過來的數據就是 e1 和 e2 兩條),之后再將 Barrier 往下廣播,當此 Barrier 傳遞到 sink 算子,snapshot 就算是制作完成了。此時 source 中還會源源不斷的產生數據,并產生新的 checkpoint ,但是此時如果 Container 宕機重啟就需要進行數據的恢復了。剛剛完成的 checkpoint 中 offset為2,count為2,那我們就按照這個 state 進行恢復。此時 Task1 會從 e3 開始消費,這就是 Recover 操作。■ checkpoint 的注意事項
下面列舉的3個注意要點都會影響到系統的吞吐,在實際開發過程中需要注意:3.背壓的產生及 Flink 的反壓處理
在分布式系統中經常會出現多個 Task 多個 JVM 之間可能需要做數據的交換,我們使用生產者和消費者來說明這個事情。假設我現在的 Producer 是使用了無界 buffer 來進行存儲,當我們的生產者生產速度遠大于消費者消費的速度時,生產端的數據會因為消費端的消費能力低下而導致數據積壓,最終導致 OOM 的產生。而就算使用了有界 buffer,同樣消費者端的消費能力低下,當 buffer 被積滿時生產者就會停止生產,這樣還不能完全地解決我們的問題,所以就需要根據不同的情況進行調整。Flink 也是通過有界 buffer 來進行不同 TaskManager 的數據交換。而且做法分為了靜態控流和動態控流兩種方式。簡單來說就是當生產者比消費者的 TPS 多時,我們采用溢寫的方式,使用 batch 來封裝好我們的數據,然后分批發送出去,每次發送完成后再 sleep 一段時間,這個時間的計算方式是 left(剩余的數據)/ tps,但是這個做法是很難去預估系統的情況的。Flink 1.5 之前的流控是基于 TCP 的滑動窗口實現的,在之前的課程中已經有提到過了。而 Flink 在1.5之后已經棄用了該機制,所以這里不展開說明。在此網絡模型中,數據生成節點只能通過檢查當前的 channel 是否可寫來決定自己是否要向消費端發送數據,它對下游數據消費端的真實容量情況一概不知。這就導致,當生成節點發現 channel 已經不可寫的時候,有可能下游消費節點已經積壓了很多數據。Credit-Based 我們用下面的數據交換的例子說明:Flink 的數據交換大致分為三種,一種是同一個 Task 的數據交換,另一種是 不同 Task 同 JVM 下的數據交換。第三種就是不同 Task 且不同 JVM 之間的交換。同一個 Task 的數據交換就是我們剛剛提到的 forward strategy 方式,主要就是避免了序列化和網絡的開銷。第二種數據交換的方式就是數據會先通過一個 record Writer ,數據在里面進行序列化之后再傳遞給 Result Partition ,之后數據會通過 local channel 傳遞給另外一個 Task 的 Input Gate 里面,再進行反序列化,推送給 Record Reader 之后進行操作。因為第三種數據交換涉及到了不同的 JVM,所以會有一定的網絡開銷,和第二種的區別就在于它先推給了 Netty ,通過netty把數據推送到遠程端的 Task 上。■ Credit-Based
此時我們可以看到 event1 已經連帶一個 backlog = 1 推送給了 TaskB,backlog 的作用其實只是為了讓消費端感知到我們生產端的情況此時 event1 被 TaskB 接收后,TaskB會返回一個 ack 給 TaskA,同時返回一個credit = 3,這個是告知 TaskA 它還能接收多少條數據,Flink 就是通過這種互相告知的方式,來讓生產者和消費者都能感知到對方的狀態。此時經過一段時間之后,TaskB中的有界 buffer 已經滿了,此時 TaskB回復 credit = 0 給 TaskA,此時 channel 通道將會停止工作,TaskA 不再將數據發往 TaskB。此時再經過一段時間,TaskA 中的有界 Buffer 也已經出現了數據積壓,所以我們平時遇到的吞吐下降,處理延遲的問題,就是因為此時整個系統相當于一個停滯的狀態,如圖二示,所有的過程都被打上 “X”,表示這些過程都已經停止工作。JVM 是一個非常復雜的系統,當其內存不足時會造成 OOM ,導致系統的崩潰。Flink 在拿到我們分配的內存之后會先分配一個 cutoff 預留內存,保證系統的安全性。Netword buffers 其實就是對應我們剛剛一直提到的有界 buffer,momery manager 是一個內存池,這部分的內存可以設置為堆內或者堆外的內存,當然在流式作業中我們一般設置其為堆外內存,而 Free 部分就是提供給用戶使用的內存塊。現在我們假設分配給此 TaskManager 的內存是 8g。- 首先是要砍掉 cutoff 的部分,默認是0.25,所以我們的可用內存就是 8gx0.75
- network buffers 占用可用內存的 0.1 ,所以是 6144x0.1
- 堆內/堆外內存為可用內存減去 network buffers 的部分,再乘以 0.8
- 給到用戶使用的內存就是堆內存剩下的 0.2 那部分
其實真實情況是 Flink 是先知道了 heap 內存的大小然后逆推出其它內存的大小。1.問題定位口訣
一壓是指背壓,遇到問題先看背壓的情況,二查就是指 checkpoint ,對齊數據的時間是否很長,state 是否很大,這些都是和系統吞吐密切相關的,三指標就是指 Flink UI 那塊的一些展示,我們的主要關注點其實就是延遲和吞吐,系統資源,還有就是 GC logs。看反壓:通常最后一個被壓高的 subTask 的下游就是 job 的瓶頸之一。
看 Checkpoint 時長:Checkpoint 時長能在一定程度影響 job 的整體吞吐。
看核心指標:指標是對一個任務性能精準判斷的依據,延遲指標和吞吐則是其中最為關鍵的指標。
資源的使用率:提高資源的利用率是最終的目的。
■ 常見的性能問題
- 在關注背壓的時候大家往往忽略了數據的序列化和反序列化過程所造成的性能問題。
- 一些數據結構,比如 HashMap 和 HashSet 這種 key 需要經過 hash 計算的數據結構,在數據量大的時候使用 keyby 進行操作, 造成的性能影響是非常大的。
- 如果我們的下游是 MySQL,HBase 這種,我們都會進行一個批處理的操作,就是讓數據存儲到一個 buffer 里面,在達到某些條件的時候再進行發送,這樣做的目的就是減少和外部系統的交互,降低網絡開銷的成本。
- 頻繁 GC ,無論是 CMS 也好,G1 也好,在進行 GC 的時候,都會停止整個作業的運行,GC 時間較長還會導致 JobManager 和 TaskManager 沒有辦法準時發送心跳,此時 JobManager 就會認為此 TaskManager 失聯,它就會另外開啟一個新的 TaskManager
- 窗口是一種可以把無限數據切割為有限數據塊的手段。比如我們知道,使用滑動窗口的時候數據的重疊問題,size = 5min 雖然不屬于大窗口的范疇,可是 step = 1s 代表1秒就要進行一次數據的處理,這樣就會造成數據的重疊很高,數據量很大的問題。
2.Flink 作業調優
我們可以通過一些數據結構,比如 Set 或者 Map 來結合 Flink state 進行去重。但是這些去重方案會隨著數據量不斷增大,從而導致性能的急劇下降,比如剛剛我們分析過的 hash 沖突帶來的寫入性能問題,內存過大導致的 GC 問題,TaskManger 的失聯問題。方案二和方案三也都是通過一些數據結構的手段去進行去重,有興趣的同學可以自行下去了解,在這里不再展開。■ 數據傾斜
數據傾斜是大家都會遇到的高頻問題,解決的方案也不少。第一種場景是當我們的并發度設置的比分區數要低時,就會造成上面所說的消費不均勻的情況。第二種提到的就是 key 分布不均勻的情況,可以通過添加隨機前綴打散它們的分布,使得數據不會集中在幾個 Task 中。在每個節點本地對相同的 key 進行一次聚合操作,類似于 MapReduce 中的本地 combiner。map-side 預聚合之后,每個節點本地就只會有一條相同的 key,因為多條相同的 key 都被聚合起來了。其他節點在拉取所有節點上的相同 key 時,就會大大減少需要拉取的數據數量,從而也就減少了磁盤 IO 以及網絡傳輸開銷。■ 內存調優
Flink 的內存結構剛剛我們已經提及到了,所以我們清楚,調優的方面主要是針對 非堆內存 Network buffer ,manager pool 和堆內存的調優,這些基本都是通過參數來進行控制的。這些參數我們都需要結合自身的情況去進行調整,這里只給出一些建議。而且對于 ManagerBuffer 來說,Flink 的流式作業現在并沒有過多使用到這部分的內存,所以我們都會設置得比較小,不超過0.3。堆內存的調優是關于 JVM 方面的,主要就是將默認使用的垃圾回收器改為 G1 ,因為默認使用的 Parallel Scavenge 對于老年代的 GC 存在一個串行化的問題,它的 Full GC 耗時較長,下面是關于 G1 的一些介紹,網上資料也非常多,這里就不展開說明了。看完上述內容,你們對如何進行Flink作業問題分析和調優實踐有進一步的了解嗎?如果還想了解更多知識或者相關內容,請關注創新互聯行業資訊頻道,感謝大家的支持。
文章標題:如何進行Flink作業問題分析和調優實踐
網站地址:http://m.newbst.com/article42/gegjec.html
成都網站建設公司_創新互聯,為您提供Google、網站營銷、網站維護、動態網站、外貿網站建設、做網站
廣告
聲明:本網站發布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網站立場,如需處理請聯系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經允許不得轉載,或轉載時需注明來源:
創新互聯