一開(kāi)始，我們采用了 JStorm 和 Spark Streaming 進(jìn)行實(shí)時(shí)開(kāi)發(fā)。這兩個(gè)計(jì)算引擎在滿(mǎn)足大部分的場(chǎng)景時(shí)，沒(méi)有特別大的問(wèn)題，但是在應(yīng)用到供應(yīng)鏈或者物流場(chǎng)景時(shí)，不是那么簡(jiǎn)單。所以我們?cè)?7年切換到了 Flink 上，F(xiàn)link 提供了一些很實(shí)用的功能，而這些功能在一些供應(yīng)鏈的場(chǎng)景下比較適用。

首先是內(nèi)部 Flink 支持一套完整的 SQL 寫(xiě)法，提高了開(kāi)發(fā)效率。第二點(diǎn)是 Flink 內(nèi)置的基于 State 的 retraction 機(jī)制，很好的支持了供應(yīng)鏈當(dāng)中的取消訂單，換配等這種操作，而且非常簡(jiǎn)單。Flink 的 CEP 功能，方便的實(shí)現(xiàn)了超時(shí)統(tǒng)計(jì)的功能，還有目前正在推的 AtoScaling 方案，比如數(shù)據(jù)傾斜，資源配置等。另外一點(diǎn)是 Flink 在批流混合問(wèn)題的處理，有很好的支持。

1. 神奇的 Retraction

左邊有4列數(shù)據(jù)，第1列數(shù)據(jù)是物流訂單，第2列的數(shù)據(jù)是物流訂單的創(chuàng)建時(shí)間，第3列數(shù)據(jù)是這個(gè)訂單是不是被取消，第4個(gè)數(shù)據(jù)是計(jì)劃配送公司，就是訂單分配給哪個(gè)配送公司。這個(gè)業(yè)務(wù)需求看上去非常簡(jiǎn)單，其實(shí)就是統(tǒng)計(jì)每個(gè)配送公司計(jì)劃履行的有效單量有多少。但是有兩個(gè)點(diǎn)需要注意一下，第一點(diǎn)，有一個(gè)訂單 LP3，在開(kāi)始的時(shí)候是有效的，然后在最后一條時(shí)取消了，就變成了一個(gè)無(wú)效訂單，但這一條訂單不應(yīng)該統(tǒng)計(jì)在內(nèi)，因?yàn)闃I(yè)務(wù)需求統(tǒng)計(jì)的是有效訂單。第二點(diǎn)，配送公司的轉(zhuǎn)變，LP1 這個(gè)訂單在一分鐘時(shí)是 tmsA 來(lái)配送，后來(lái)變成了 tmsB，再過(guò)了一段時(shí)間，這個(gè)訂單又變成 tmsC 配送。如果基于 Storm 增量計(jì)算，得出的結(jié)果顯然是錯(cuò)誤的，這時(shí)要按照最后一次消息給我們傳過(guò)來(lái)的數(shù)據(jù)統(tǒng)計(jì)。這樣的場(chǎng)景在 Flink 中是如何實(shí)現(xiàn)的呢？Flink 也提供了一些非常好的回撤機(jī)制。

第一段代碼使用 Flink 的 last_value 函數(shù)，獲取這個(gè)訂單最后一個(gè)消息非空的值，在這個(gè)基礎(chǔ)上進(jìn)行匯總。一旦 last_value 中字段發(fā)生變化，都會(huì)觸發(fā)撤回機(jī)制，得到最后正確的值。

2. 實(shí)時(shí)超時(shí)統(tǒng)計(jì)

這個(gè)案例發(fā)生在菜鳥(niǎo)實(shí)際物流場(chǎng)景中，第1個(gè)表格是一個(gè)日志的時(shí)間，第2個(gè)是物流訂單，第3個(gè)字段是出庫(kù)時(shí)間，最后一個(gè)字段是攬收時(shí)間，現(xiàn)在需要統(tǒng)計(jì)的是出庫(kù)超6個(gè)小時(shí)沒(méi)有被攬收的單量。這里涉及到全鏈路時(shí)效問(wèn)題，全鏈路時(shí)效指從下單到倉(cāng)庫(kù)發(fā)貨到快遞攬收到簽收的整體時(shí)間，這個(gè)場(chǎng)景放在離線中是非常容易實(shí)現(xiàn)的，但是放到實(shí)時(shí)中來(lái)不是很簡(jiǎn)單， 比如 LP1 在00:05分的時(shí)候沒(méi)有攬收，現(xiàn)在如果當(dāng)下時(shí)刻是12點(diǎn)的話，也沒(méi)有攬收，理論上應(yīng)該計(jì)算這個(gè)訂單，但是沒(méi)有攬收就意味著沒(méi)有消息進(jìn)來(lái)，沒(méi)有消息進(jìn)來(lái)我們又要統(tǒng)計(jì)，其實(shí)我們用 Flink 是沒(méi)法統(tǒng)計(jì)的，那這種情況我們?cè)趺刺幚砟?？我們的解決方案是如果沒(méi)有這條消息我們用 Flink 來(lái)制造這條消息。這種超時(shí)消息統(tǒng)計(jì)我們想到了幾種方法：

包括引入消息中間件 ( kafka ) 和 Flink 的 CEP。最終選擇了 Flink 的 Timer Service，因?yàn)檫@種消息不是特別多，中間件又特別重。而 CEP 會(huì)丟掉一些回傳不準(zhǔn)確的消息，導(dǎo)致數(shù)據(jù)計(jì)算不準(zhǔn)確，針對(duì)這些情況，我們?cè)谡{(diào)研之后選擇了 Timer Service，同時(shí)我們對(duì)它底層的 ProcessElement 和 OnTimer 兩個(gè)方法進(jìn)行了改寫(xiě)。ProcessElement 告訴 Flink 存儲(chǔ)什么樣的數(shù)據(jù)，然后啟動(dòng)針對(duì)每一個(gè)超時(shí)的事件的 Timer Service。OnTimer 方法會(huì)在每個(gè)超時(shí)的時(shí)刻讀這個(gè)超時(shí)的消息，并把這個(gè)超時(shí)的消息下發(fā)下來(lái)?；谙掠胃Ａ鞯年P(guān)聯(lián)操作之后就能計(jì)算超時(shí)消息的單量。

先構(gòu)造一個(gè) process funcation 到 state 存數(shù)據(jù)，并為每一個(gè)超時(shí)的數(shù)據(jù)注冊(cè)一個(gè) Timer Service。然后執(zhí)行 OnTimer 這個(gè)方法，讀取并把這個(gè)超時(shí)的消息下發(fā)下去。

3. 從手動(dòng)優(yōu)化到智能優(yōu)化

關(guān)于數(shù)據(jù)傾斜的問(wèn)題，左圖顯示在 map 階段 shuffer 之后數(shù)據(jù)傾斜到了紅色的 Agg 上，這時(shí)就出現(xiàn)熱點(diǎn)了，原來(lái)我們是對(duì)這個(gè) lg_order_code 進(jìn)行 hash 取值操作，然后再針對(duì)散列的結(jié)果進(jìn)行二次的聚合，這樣操作后在一定程度上減輕了數(shù)據(jù)的傾斜。在最近的 Flink 的版本中已經(jīng)實(shí)現(xiàn)了規(guī)避數(shù)據(jù)傾斜的方法，我們內(nèi)部的 Blink 版本，有幾個(gè)功能去優(yōu)化熱點(diǎn)的問(wèn)題，第一個(gè)就是 MiniBatch，之前來(lái)一條數(shù)據(jù)，我們就去 State 里面查詢(xún)?nèi)缓髮?xiě)入，MiniBatch 的作用是把所有的數(shù)據(jù)先聚合一次，類(lèi)似一個(gè)微批處理，然后再把這個(gè)數(shù)據(jù)寫(xiě)到 State 里面，或者在從 State 里面查出來(lái)，這樣可以大大的減輕對(duì) State 查詢(xún)的壓力。第二個(gè)辦法就是 LocalGlobal，類(lèi)似于在 hive 中 map 階段中的 combiner，通過(guò)設(shè)置這個(gè)參數(shù)可以在讀的時(shí)候先聚合。第三個(gè)辦法是 PartialFinal，類(lèi)似于散列的方式，分兩次聚合，相當(dāng)于 hive 中兩個(gè)入 reduce 操作。通過(guò)設(shè)置這三個(gè)參數(shù)，可以在大部分場(chǎng)景規(guī)避數(shù)據(jù)傾斜的問(wèn)題。

智能化功能支持的另一個(gè)場(chǎng)景是資源配置。在進(jìn)行實(shí)時(shí) ETL 過(guò)程中，首先要定義 DDL，然后編寫(xiě) SQL，之后需要進(jìn)行資源配置。針對(duì)資源配置問(wèn)題，菜鳥(niǎo)之前的方案是對(duì)每一個(gè)節(jié)點(diǎn)進(jìn)行配置，包括并發(fā)量、是否會(huì)涉及消息亂序操作、CPU、內(nèi)存等，一方面配置過(guò)程非常復(fù)雜，另一方面無(wú)法提前預(yù)知某些節(jié)點(diǎn)的資源消耗量。Flink 目前提供了較好的優(yōu)化方案來(lái)解決該問(wèn)題：

• 大促場(chǎng)景：該場(chǎng)景下，菜鳥(niǎo)會(huì)提前預(yù)估該場(chǎng)景下的 QPS，會(huì)將其配置到作業(yè)中并重啟。重啟后 Flink 會(huì)自動(dòng)進(jìn)行壓測(cè)，測(cè)試該 QPS 每個(gè)節(jié)點(diǎn)所需要的資源。

• 日常場(chǎng)景：日常場(chǎng)景的 QPS 峰值可能遠(yuǎn)遠(yuǎn)小于大促場(chǎng)景，此時(shí)逐一配置 QPS 依然會(huì)很復(fù)雜。為此 Flink 提供了 AutoScaling 智能調(diào)優(yōu)的功能，除了可以支持大促場(chǎng)景下提前設(shè)置 QPS 并壓測(cè)獲取所需資源，還可以根據(jù)上游下發(fā)的 QPS 的數(shù)據(jù)自動(dòng)預(yù)估需要的資源。大大簡(jiǎn)化了資源配置的復(fù)雜度，使得開(kāi)發(fā)人員可以更好地關(guān)注業(yè)務(wù)邏輯本身的開(kāi)發(fā)。

特別推薦一個(gè)分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒(méi)關(guān)注的小伙伴，可以長(zhǎng)按關(guān)注一下：
																																																																													

																																																																														

長(zhǎng)按訂閱更多精彩▼
																																																																													

																																																																														
如有收獲，點(diǎn)個(gè)在看，誠(chéng)摯感謝