一開始，我們采用了 JStorm 和 Spark Streaming 進行實時開發(fā)。這兩個計算引擎在滿足大部分的場景時，沒有特別大的問題，但是在應用到供應鏈或者物流場景時，不是那么簡單。所以我們在17年切換到了 Flink 上，F(xiàn)link 提供了一些很實用的功能，而這些功能在一些供應鏈的場景下比較適用。

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

1. 神奇的 Retraction

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

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

2. 實時超時統(tǒng)計

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

包括引入消息中間件 ( kafka ) 和 Flink 的 CEP。最終選擇了 Flink 的 Timer Service，因為這種消息不是特別多，中間件又特別重。而 CEP 會丟掉一些回傳不準確的消息，導致數(shù)據(jù)計算不準確，針對這些情況，我們在調(diào)研之后選擇了 Timer Service，同時我們對它底層的 ProcessElement 和 OnTimer 兩個方法進行了改寫。ProcessElement 告訴 Flink 存儲什么樣的數(shù)據(jù)，然后啟動針對每一個超時的事件的 Timer Service。OnTimer 方法會在每個超時的時刻讀這個超時的消息，并把這個超時的消息下發(fā)下來。基于下游跟正常流的關聯(lián)操作之后就能計算超時消息的單量。

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

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

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

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

• 大促場景：該場景下，菜鳥會提前預估該場景下的 QPS，會將其配置到作業(yè)中并重啟。重啟后 Flink 會自動進行壓測，測試該 QPS 每個節(jié)點所需要的資源。

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

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

																																																																														

長按訂閱更多精彩▼
																																																																													

																																																																														
如有收獲，點個在看，誠摯感謝