Flink水位線Watermaker生產(chǎn)應(yīng)用避坑分享

IT那活兒發(fā)布于2023-01-11 13:20 / 3936人閱讀

基于對(duì)處理時(shí)間的實(shí)時(shí)性要求，很多業(yè)務(wù)場景對(duì)“延遲”的忍受能力越來越低，因?yàn)槟芨皶r(shí)發(fā)現(xiàn)問題，就能及時(shí)解決問題，進(jìn)而能提升支撐保證和體驗(yàn)。在大數(shù)據(jù)分析領(lǐng)域，數(shù)據(jù)分析地越及時(shí)，價(jià)值越高，尤其是在推薦、風(fēng)控等場景中，對(duì)實(shí)時(shí)性的要求更為苛刻。而流計(jì)算天然支持對(duì)事件發(fā)生的先后順序、時(shí)間關(guān)系方面的分析，這也是flink在實(shí)時(shí)場景及大數(shù)據(jù)中被越來越多使用的原因。今天就和大家分享一下Flink的時(shí)間機(jī)制介紹，及展開一下watermark。

Flink三種時(shí)間機(jī)制

Processing Time: 機(jī)器或者系統(tǒng)的時(shí)間，可理解為真實(shí)世界的時(shí)間。使用該時(shí)間模式有最好的性能和最低的延遲。
Event time: 數(shù)據(jù)上自帶的時(shí)間，可理解為數(shù)據(jù)世界的時(shí)間。實(shí)際場景中應(yīng)用較多，由于數(shù)據(jù)在傳輸過程有網(wǎng)絡(luò)、I/O以及消費(fèi)等因素，往往不能保證數(shù)據(jù)按順序到達(dá)，因此導(dǎo)致了時(shí)間的亂序等問題。
Ingestion time: 數(shù)據(jù)進(jìn)入程序時(shí)的時(shí)間，比如12點(diǎn)的一條數(shù)據(jù)與11點(diǎn)的一條數(shù)據(jù)同時(shí)進(jìn)入程序，這兩者會(huì)被認(rèn)為是同一時(shí)間的數(shù)據(jù)。與事件時(shí)間相比，攝入時(shí)間程序不能處理任何無序事件或者延遲事件，但是程序無需指定如何產(chǎn)生水印。

最常用的兩種時(shí)間ProcessingTime和Eventtime對(duì)比：

PS：對(duì)時(shí)間的理解，時(shí)間并不一定就一定是時(shí)間，只要數(shù)據(jù)是有序遞增的，都可以理解為時(shí)間來進(jìn)行處理。

定義使用何種時(shí)間機(jī)制：

Watermaker的理解

在實(shí)際業(yè)務(wù)場景中的實(shí)時(shí)計(jì)算，往往都是使用的數(shù)據(jù)時(shí)間EventTime，這樣才能保證數(shù)據(jù)的真實(shí)性和準(zhǔn)確性。但是數(shù)據(jù)在傳輸過程有網(wǎng)絡(luò)、I/O以及消費(fèi)等因素，數(shù)據(jù)的時(shí)間可能會(huì)存在一定程度的亂序。

需要考慮對(duì)于整個(gè)序列進(jìn)行更大程度離散化。把數(shù)據(jù)按照一定的條數(shù)組成一些小批次，但這里的小批次并不是攢夠多少條就要去處理，而是為了對(duì)他們進(jìn)行時(shí)間上的劃分。

經(jīng)過這種更高層次的離散化之后，我們會(huì)發(fā)現(xiàn)最右邊方框里的時(shí)間就是一定會(huì)小于中間方框里的時(shí)間，中間框里的時(shí)間也一定會(huì)小于最左邊方框里的時(shí)間。

這個(gè)時(shí)候我們?cè)谡麄€(gè)時(shí)間序列里插入一些類似于標(biāo)志位的一些特殊的處理數(shù)據(jù)，這些特殊的處理數(shù)據(jù)叫做watermark。一個(gè)watermark 本質(zhì)上就代表了這個(gè)watermark 所包含的timestamp數(shù)值，表示以后到來的數(shù)據(jù)已經(jīng)再也沒有小于或等于這個(gè)時(shí)間的了。

Watermaker的特性

watermark 會(huì)以廣播的形式在算子之間進(jìn)行傳播，下游所有算子共享watermark。
如果在程序里面收到了一個(gè) Long.MAX_VALUE 這個(gè)數(shù)值的 watermark，就表示對(duì)應(yīng)的那一條流的一個(gè)部分不會(huì)再有數(shù)據(jù)發(fā)過來了，它相當(dāng)于就是一個(gè)終止的一個(gè)標(biāo)志。
對(duì)于單流而言，會(huì)選擇當(dāng)前最大的值timestamp作為watermark。對(duì)于多流而言，會(huì)選擇流中最小的watermark作為整個(gè)任務(wù)的watermark。即可看做一個(gè)由多個(gè)木塊組成的裝水的木桶，桶里面水多高取決于組成桶的那個(gè)最低的木塊。

Watermaker的使用技巧

Watermaker的生成有兩類。第一類是定期生成器，默認(rèn)50ms向下游發(fā)送一次；第二類是根據(jù)一些在流處理數(shù)據(jù)流中遇到的一些特殊記錄生成的，來一條數(shù)據(jù)獲取一次，發(fā)送一次。生產(chǎn)中的使用可根據(jù)業(yè)務(wù)考慮使用何種，已達(dá)到性能和業(yè)務(wù)的平衡。

關(guān)于數(shù)據(jù)的延遲亂序，生成Watermaker時(shí)是可以直接增加一個(gè)特定延遲時(shí)間的。這樣做的好處是，在水位到達(dá)時(shí)，仍然可以再等待一個(gè)延遲保證晚到的數(shù)據(jù)進(jìn)行統(tǒng)計(jì)，保證數(shù)據(jù)的準(zhǔn)確性，當(dāng)然這樣也使得數(shù)據(jù)實(shí)時(shí)性延遲，是保證實(shí)時(shí)性還是準(zhǔn)確性，需要生成進(jìn)行取舍，或者兩種之間采用一個(gè)平衡值。具體的延遲時(shí)長，需要觀察實(shí)際數(shù)據(jù)的延遲等進(jìn)行判斷及定義。

Watermaker實(shí)際應(yīng)用避坑指南

一、防止數(shù)據(jù)傾斜，使用Watermaker沒有觸發(fā)數(shù)據(jù)匯總

場景：

數(shù)據(jù)源一分鐘產(chǎn)生一條數(shù)據(jù)，每條數(shù)據(jù)中有9條左右的不同key的子數(shù)據(jù)，程序進(jìn)行Keyby處理后，開啟一分鐘的窗口進(jìn)行匯總統(tǒng)計(jì)數(shù)量。

問題：

程序啟動(dòng)4個(gè)并行進(jìn)行處理，結(jié)果幾分鐘后都沒觸發(fā)匯總。什么原因？

原因：通過前臺(tái)對(duì)flink任務(wù)的監(jiān)控發(fā)現(xiàn)，4個(gè)并行后由于數(shù)據(jù)量太少，有一個(gè)并行沒有收到數(shù)據(jù)，因此沒有產(chǎn)生Watermaker，由Watermaker的特性的第三條可以理解，整個(gè)程序目前的watermarker取的是第4個(gè)并行的watermarker初始值Long.MIN_VALUE，所以導(dǎo)致整個(gè)程序沒有進(jìn)行觸發(fā)匯總。

如何解決：

不改并行的情況下，需要對(duì)程序Watermaker生成之前進(jìn)行數(shù)據(jù)負(fù)載均衡，最簡單直接的辦法是進(jìn)行一次keyby處理。
數(shù)據(jù)量較少的情況，直接改小并行度。

兩種方法的目的都是保證每個(gè)并行都能消費(fèi)到實(shí)時(shí)數(shù)據(jù)，這里我們采用第一個(gè)方案進(jìn)行修改驗(yàn)證，結(jié)果如圖時(shí)間小于1593572813000的數(shù)據(jù)都會(huì)及時(shí)進(jìn)行匯總生成指標(biāo)。

實(shí)際生產(chǎn)中關(guān)于數(shù)據(jù)負(fù)載均衡的問題往往也是需要注意的，往往數(shù)據(jù)的傾斜問題，如果比較嚴(yán)重會(huì)導(dǎo)致數(shù)據(jù)計(jì)算的準(zhǔn)確性以及整個(gè)任務(wù)的性能等一系列問題，關(guān)于數(shù)據(jù)傾斜問題這里不進(jìn)行深入探討，下期有機(jī)會(huì)給大家做進(jìn)一步的分享。

二、業(yè)務(wù)鏈實(shí)時(shí)指標(biāo)計(jì)算延遲問題

場景：業(yè)務(wù)鏈實(shí)時(shí)指標(biāo)計(jì)算延遲。

原因：重復(fù)注冊(cè)Watermaker導(dǎo)致任務(wù)吞吐量變低，影響計(jì)算效率。

如何解決：

業(yè)務(wù)鏈處理經(jīng)過算子處理之后m條數(shù)據(jù)會(huì)生成m*n條數(shù)據(jù)，然后進(jìn)行keyby匯總。之前水位注冊(cè)在匯總數(shù)據(jù)之前，因此需要對(duì)m*n條數(shù)據(jù)都進(jìn)行水位注冊(cè)，使得同一時(shí)間多次水位處理，程序效率也下來了，整個(gè)任務(wù)吞吐量變低。利用水位廣播傳遞的特點(diǎn)，將水位注冊(cè)放到數(shù)據(jù)源，只需要對(duì)m條數(shù)據(jù)進(jìn)行注冊(cè)，處理邏輯直接少了n倍，整個(gè)任務(wù)吞吐量也隨之上來了

建議生成Watermaker的工作越靠近DataSource越好。這樣會(huì)方便讓程序邏輯里面更多的operator去判斷某些數(shù)據(jù)是否亂序。Flink內(nèi)部提供了很好的機(jī)制去保證這些timestamp和watermark被正確地傳遞到下游的節(jié)點(diǎn)。

今天分享到此結(jié)束，后頭見。