背景

菜鳥從2017年4月開始探索 Blink（即 Apache Flink 的阿里內(nèi)部版本），2017年7月開始在線上環(huán)境使用 Blink，作為我們的主流實時計算引擎。

為什么短短幾個月的探索之后，我們就選擇Blink作為我們主要的實時計算引擎呢？

在效率上，Blink 提供 DataStream、TableAPI、SQL 三種開發(fā)模式，強大的 SQL 模式已經(jīng)滿足大部分業(yè)務場景，配合半智能資源優(yōu)化、智能傾斜優(yōu)化、智能作業(yè)壓測等功能，可以極大地提升實時作業(yè)的開發(fā)效率；在性能上，諸如MiniBatch&MicroBatch、維表 Async&Cache、利用 Niagara 進行本地狀態(tài)管理等內(nèi)部優(yōu)化方案，可以極大地提升實時作業(yè)的性能；在保障上，Blink 自帶的 Failover 恢復機制，能夠?qū)崿F(xiàn)線程級的恢復，可以做到分鐘級恢復，配合 Kmonitor 監(jiān)控平臺、烽火臺預警平臺，可以有效地實現(xiàn)實時作業(yè)的數(shù)據(jù)保障。

接下來，我將結(jié)合供應鏈業(yè)務的一些業(yè)務場景，簡要說明，Blink 如何解決我們遇到的一些實際問題。

回撤機制

訂單履行是供應鏈業(yè)務中最常見的物流場景。什么是訂單履行呢？當商家 ERP 推單給菜鳥之后，菜鳥履行系統(tǒng)會實時計算出每筆訂單的出庫、攬收、簽收等節(jié)點的預計時間，配送公司需要按照各節(jié)點的預計時間進行訂單的配送。為了保證訂單的準點履約，我們經(jīng)常需要統(tǒng)計每家配送公司每天各個節(jié)點的預計單量，便于配送公司提前準備產(chǎn)能。

看似很簡單的實時統(tǒng)計加工，我們在開發(fā)過程中遇到了什么問題呢？履行重算！當物流訂單的上游某個節(jié)點延遲時，履行系統(tǒng)會自動重算該筆訂單下游所有節(jié)點的預計時間。比如某個物流訂單出庫晚點后，其后的預計攬收時間、預計簽收時間都會重算。而對于大部分的實時計算引擎來說，并不能很友好的支持這種變 Key 統(tǒng)計的問題。以前，數(shù)據(jù)量沒那么大的時候，還可以通過 OLAP 數(shù)據(jù)庫來解決這類場景，當量上來后， OLAP 方案的成本、性能都是很大的問題。

除了 OLAP 方案，我們提倡采用 Blink 已經(jīng)內(nèi)置的 Retraction 機制，來解決這類變 Key 統(tǒng)計的問題，這也是我們在2017年初就開始嘗試 Blink 的重要原因。Blink 的Retraction 機制，使用 State 在內(nèi)存或者外部存儲設備中對數(shù)據(jù)進行統(tǒng)計處理，當上游數(shù)據(jù)源對某些匯總 Key 的數(shù)據(jù)做更新時，Blink 會主動給下游下發(fā)一個刪除消息從而“撤回”之前的那條消息，并用最新下發(fā)的消息對表做更新操作。

下面是一個簡化后的案例，供了解Blink Retraction的內(nèi)部計算過程：

對于上述案例，可以通過 Blink 提供的強大的、靈活的、簡易的 SQL 開發(fā)模式來實現(xiàn)，只需要幾行 SQL 即可完成。

select plan_tms_sign_time ,sum(1) as plan_tms_sign_lgtord_cntfrom (select lg_order_code ,last_value(plan_tms_sign_time) as plan_tms_sign_time from dwd_csn_whc_lgt_fl_ord_ri group by lg_order_code ) ssgroup by plan_tms_sign_time;

維表關(guān)聯(lián)

供應鏈業(yè)務的實體角色非常多（倉、配、分撥、站點、小件員、貨主、行業(yè)、地區(qū)等），實體繁多，這意味著我們在建設實時明細中間層的時候，會使用大量的維表關(guān)聯(lián)，這對 Blink 在維表關(guān)聯(lián)的性能上提出了更高的要求如何提升大量的大小維表的關(guān)聯(lián)性能？Blink 從來沒讓用戶失望，Blink SQL 模式在維表關(guān)聯(lián)的性能上，也做了大量的優(yōu)化：

優(yōu)化1：Async IO，有一些實時計算引擎，維表關(guān)聯(lián)是采用同步訪問的方式，即來一條數(shù)據(jù)，去數(shù)據(jù)庫查詢一次，等待返回后輸出關(guān)聯(lián)結(jié)果。這種方式，可以發(fā)現(xiàn)網(wǎng)絡等待時間極大地阻礙了吞吐和延遲。而 Blink 采用了異步訪問的模式，可以并發(fā)地處理多個請求和回復，從而連續(xù)地請求之間不需要阻塞等待，吞吐量大大提升。

優(yōu)化2：緩存，維表關(guān)聯(lián)涉及到大量的維表查詢請求，其中可能存在大量相同 Key 的重復請求。Blink SQL 模式提供了緩存的機制，并提供 LRU 和 ALLCache 兩種緩存方案。

用戶可以通過配置 Cache='LRU' 參數(shù)，開啟 LRU 緩存優(yōu)化。開啟后，Blink 會為每個 JoinTable 節(jié)點創(chuàng)建一個 LRU 本地緩存。當每個查詢進來的時候，先去緩存中查詢，如果存在則直接關(guān)聯(lián)輸出，減少了一次 IO 請求。如果不存在，再發(fā)起數(shù)據(jù)庫查詢請求，請求返回的結(jié)果會先存入緩存中以備下次查詢。

如果維表數(shù)據(jù)不大，用戶可以通過配置 Cache='ALL' 參數(shù)，對維表進行全量緩存。這樣，所有對該維表的查詢操作，都會直接走本地緩存模式，幾乎沒有 IO，關(guān)聯(lián)的性能非常好。

優(yōu)化3：緩存無效 Key，如果維表很大，無法采用 ALLCache 的方案，而在使用 LRU 緩存時，會存在不少維表中不存在的 Key 。由于命中不了緩存，導致緩存的收益較低，仍然會有大量請求發(fā)送到數(shù)據(jù)庫，并且LRU模式下緩存里的key不會永久保留，可以通過調(diào)整參數(shù)，設置保留時間。

優(yōu)化4：Distribute By 提高緩存命中率，默認情況下，維表關(guān)聯(lián)的節(jié)點與上游節(jié)點之間是 Chain 在一起，不經(jīng)過網(wǎng)絡。這在緩存大小有限、Key 總量大、熱點不明顯的情況下, 緩存的收益可能較低。這種情況下可以將上游節(jié)點與維表關(guān)聯(lián)節(jié)點的數(shù)據(jù)傳輸改成按 Key 分區(qū)。這樣通?？梢钥s小單個節(jié)點的 Key 個數(shù)，提高緩存的命中率。

除了上述幾點優(yōu)化，Blink SQL 模式還在嘗試引入 SideInput、Partitioned ALL Cache 等優(yōu)化方案，相信在隨后開源的 Blink 版本中，維表關(guān)聯(lián)的性能會越來越好。

下面是一張來自 Flink Committer 云邪 異步查詢的流程圖，供理解與同步請求的差異。

數(shù)據(jù)傾斜

無數(shù)據(jù)不傾斜，我們在實時數(shù)倉建設過程中，也當然會遇到數(shù)據(jù)傾斜問題。在統(tǒng)計賣家的單量時，有些賣家單量大，有些賣家單量小，單量超大的賣家，就會產(chǎn)生數(shù)據(jù)傾斜；在統(tǒng)計行業(yè)的單量時，有些行業(yè)單量大，有些行業(yè)單量小，單量超大的行業(yè)，就會產(chǎn)生數(shù)據(jù)傾斜；在統(tǒng)計貨品的庫存流水情況時，有些貨品庫存流水頻繁，一些貨品庫存流水較少，庫存流水超頻繁的貨品就會產(chǎn)生數(shù)據(jù)傾斜……

我們應該如何處理數(shù)據(jù)傾斜問題呢？以統(tǒng)計賣家的單量為例，以前我們會先把訂單這個 Key 作 Hash，先針對 Hash 之后的值做一次去重的聚合操作，再在此基礎上，再做一次針對原 Key 去重的聚合操作。兩次類似的聚合操作，導致代碼寫起來比較復雜，體力勞動比較多。

2017年，我們的實時數(shù)據(jù)開始全面切換到 Blink 上，Blink 在數(shù)據(jù)傾斜這塊，又給我們提供了什么的方案呢？Blink 給出的答案是：MiniBatch/MicroBatch+LocalGlobal+PartialFinal。

MiniBatch/MicroBatch，可以實現(xiàn)微批處理，進而減少對 State 的訪問，提升吞吐。因為微批處理會導致一定的延遲，最好結(jié)合 Blink 提供的允許延遲的相關(guān)參數(shù)來使用。

LocalGlobal，分為 Local 和 Global 兩個階段，有點類似 MapReduce 中的Combine 和 Reduce 兩個階段。LocalGlobal 可以很好地處理非去重類的聚合操作，但對 Count Distinct 的優(yōu)化效果一般，因為在 Local 階段，可能 Distinct Key的去重率并不會很高，進而導致后續(xù)的 Global 階段，仍然會有熱點。

PartialFinal，可以很好地解決 Count Distinct 帶來的數(shù)據(jù)傾斜問題。PartialFinal 可以將 Distinct Key 自動打散，先聚合一次，在此基礎上，再聚合一次，從而實現(xiàn)打散熱點的作用。PartialFinal 跟手動 Hash 再聚合兩次的效果一致，通過 Blink 提供的 PartialFinal 參數(shù)，可以自動實現(xiàn)，不再需要人為手工編寫 Hash 再聚合兩次的代碼。

由上可以看出，Blink 在數(shù)據(jù)傾斜的處理上，已經(jīng)實現(xiàn)了自動化，以前人為編寫的打散熱點方案，現(xiàn)在幾個參數(shù)就能全部搞定，大大提升了代碼的編寫效率。

下面是相關(guān)參數(shù)，用戶可以直接在 Blink 的作業(yè)參數(shù)中進行配置。

# miniBatch/microBatch攢批的間隔時間blink.miniBatch.allowLatencyMs=5000blink.microBatch.allowLatencyMs=5000# 防止OOM，每個批次最多緩存多少條數(shù)據(jù)blink.miniBatch.size=20000# 開啟LocalGlobalblink.localAgg.enabled=true# 開啟PartialFinalblink.partialAgg.enabled=true

超時統(tǒng)計

上架是倉儲業(yè)務的重要組成部分。上架，顧名思義，就是要把到倉的貨品，上到倉庫的存儲貨架上。上架一般分為采購上架、銷退上架、調(diào)撥上架等。及時上架是對倉庫的重要考核項之一，無論哪一種類型的上架，我們經(jīng)常需要針對到貨后超過 x 小時未上架的訂單進行預警。

但是，Blink 的計算是消息機制，需要上游發(fā)送消息才能觸發(fā)下游計算，而上述的場景中，未上架就說明不會有上架的消息流入 Blink，進而無法完成下游的計算。

對于這種實時超時統(tǒng)計的問題，應該如何來解呢？我們嘗試了幾種方案，供參考：

方案1：針對部分 Source Connector，Blink 提供了"延時下發(fā)"的功能，用戶可以通過指定 DataDeliveryDelayMs 參數(shù)，實現(xiàn)消息延遲下發(fā)。正常的消息正常流入，正常消息也可以通過配置該參數(shù)，使其按照自己的需求延時流入。這樣，通過正常流入的消息關(guān)聯(lián)延時流入的消息，可以觸發(fā) Blink 在消息正常流入時計算一次，在延時消息流入時再觸發(fā)計算一次。這種方案，可以實現(xiàn)我們的業(yè)務需求，但是這種方案會把所有消息重新發(fā)送一遍，而不僅僅是到貨后超過x小時未上架的消息，這樣會造成計算資源的浪費，我們不建議在數(shù)據(jù)量很大的場景下使用該方案。

方案2：如果有第三方的消息中間件，而這個消息中間件又能支持配置超時下發(fā)的規(guī)則，這將是一個比較好的方案。據(jù)了解，Kafka 的最新版本已經(jīng)能夠根據(jù)業(yè)務需求，配置消息超時下發(fā)的規(guī)則。我們只需要在 Blink 中，通過正常流入的消息流關(guān)聯(lián)關(guān)鍵Kafka 超時下發(fā)的消息流，就可以觸發(fā) Blink 進行超時消息的統(tǒng)計。這樣，除了Blink，我們需要同時保障 Kafka 的穩(wěn)定性。Kafka的超時消息訂閱，可以參見：[1]。

方案3：我們能夠很自然的想到 CEP，而 Blink 也已經(jīng)提供了 CEP 的功能，且已經(jīng)SQL化。用戶可以通過 Blink CEP 完成上述業(yè)務需求的統(tǒng)計。在實操過程中，我們發(fā)現(xiàn)，通過 Blink CEP 統(tǒng)計的結(jié)果，往往與真實結(jié)果(明細匯總統(tǒng)計)有一定的出入。什么原因呢？原來到貨時間，被回傳了多次，有可能開始回傳的是9點，但是后面發(fā)現(xiàn)回傳錯了，改成了8點，而 CEP 的 Watermark 是全局地向前走的，對于這種場景，無法很好的適配。

方案4：Flink 的 ProcessFunction，是一個 Low-Level 的流處理操作。通過改寫其中的 ProcessElement 方法，可以告訴 Blink的State 里面存什么，以及如何更新State；通過改寫 OnTimer 方法，可以告訴 State 何時下發(fā)超時消息。通過對上述幾種方案的原理對比及性能壓測，我們最終選擇的也是這套方案。由于超時場景，在供應鏈業(yè)務中非常常見，我們已經(jīng)將該方案沉淀下來，同樣的場景，通過 1min 配置下相關(guān)參數(shù)，即可完成類似場景超時消息的下發(fā)。

下面是方案4簡化后的實現(xiàn)框架圖，供了解相關(guān)實現(xiàn)及優(yōu)勢。

零點起跳

每次大促，大屏上零點時刻雙十一的零點時刻一直是大家關(guān)注的焦點，為了在零點一過就讓各項指標盡快在大屏上展現(xiàn)出來，我們進行了一些端到端的優(yōu)化，供參考。

優(yōu)化1：合理調(diào)整 Blink 讀取上游消息源的 FlushInterval 。我們知道 Blink 是以Block 的形式傳輸數(shù)據(jù)，如果 Block 一直積攢不滿，Block 可能一直等待無法下發(fā)。這種情況，我們可以通過調(diào)整 FlushInterval 參數(shù)，直接控制多長時間往下游 sink 一次。這樣，Block 積滿或間隔達到滿足其中一個條件，Block 就會往下流。

優(yōu)化2：合理調(diào)整 MiniBatch/MicroBatch的size 和 AllowLatency 參數(shù)。前文提到，MiniBatch/MicroBatch 是微批處理模式，都會帶來一定的延遲，可以通過合理控制 Size 和 AllowLatency 參數(shù)，來控制該模式帶來的延遲。與優(yōu)化1一樣，兩者滿足其一，就會往下繼續(xù)執(zhí)行。

優(yōu)化3：合理控制寫 Checkpoint 的方式以及 Checkpoint 的大小。利用 Checkpoint 實現(xiàn) Exactly Once 的容錯方式一直是 Flink 作為流引擎的一個亮點。但是過于復雜的運算和網(wǎng)絡環(huán)境有可能導致 checkpoint 的對齊時間過長，從而導致整個 Job 的延遲變長。同時，Exactly Once 模式下做 Checkpoint 的時間間隔與整個任務中數(shù)據(jù)流的延遲也是一個 Trade Off。因此我們在處理特別復雜的 Job 時也將這個因素考慮了進去，并沒有使用默認的 Exactly Once 方式，而是依舊實際需求采用了 At Least Once 。同時，將 Checkpoint 的周期設置為了60s，盡可能的保證了任務在延遲較小的情況下，在 Failover 的情形下仍然能做到快速恢復。

優(yōu)化4：除了 Blink 端，在數(shù)據(jù)服務端，大屏上的實時數(shù)據(jù)，我們建議采用查詢性能優(yōu)異的 Hbase 作為存儲引擎，可以保證零點一過，三秒內(nèi)便能實現(xiàn)大屏數(shù)據(jù)的跳動。

……

未來展望

Blink 在不斷快速地發(fā)展，不僅僅是流處理，當前也開始支持批處理，用戶只需要寫一套代碼就可以同時實現(xiàn)批和流的數(shù)據(jù)開發(fā)，當前在日志型的數(shù)據(jù)場景上，我們也正在探索利用 Blink 直接實現(xiàn)批流混合模式；不僅僅是半智能資源調(diào)優(yōu)，當前開始內(nèi)測智能資源調(diào)優(yōu)，Blink 可以根據(jù)吞吐量、算子復雜度等因素，對線上作業(yè)的資源配置進行全智能自適應調(diào)優(yōu)，再也不用在大促前手動更改資源配置；不僅僅是 Java，更期望有 Python 等多語言生態(tài)，來描述計算邏輯，相信開發(fā)效率又會上一個新的臺階；不僅僅是 ETL，更期望有更廣闊的大數(shù)據(jù)算法集成，可以實現(xiàn)復雜的大數(shù)據(jù)AI場景……未來已來，我們相信，Blink 已經(jīng)做好了迎接未來的準備。

參考資料：

[1]https://ketao1989.github.io/2016/01/02/delayed-message-consume-service-use-kafka/