菜鳥實時數(shù)倉2.0進階之路

導讀：供應鏈物流場景下的業(yè)務復雜度高，業(yè)務鏈路長，節(jié)點多，實體多，實時數(shù)倉建設難度高。菜鳥跨境進口業(yè)務場景更是如此，更復雜的場景帶來更復雜的實體數(shù)據(jù)模型，對接的業(yè)務系統(tǒng)多導致ETL流程特別復雜，還有海量的日均處理數(shù)據(jù)量，使得團隊在建設進口實時數(shù)倉的過程中，面臨著諸多挑戰(zhàn)：如何保證復雜實體關系下的數(shù)據(jù)準確性？如何降低多數(shù)據(jù)源情況下的數(shù)據(jù)處理復雜度？如何提升實時多流Join的處理效率？如何實現(xiàn)實時超時統(tǒng)計？如何實現(xiàn)異常情況下的數(shù)據(jù)狀態(tài)恢復？本文主要分享菜鳥進口實時數(shù)倉的升級經(jīng)驗，以及如何利用Flink的特性解決在開發(fā)實踐中遇到的問題。

主要內(nèi)容包括：

• 相關背景介紹

• 進口實時數(shù)倉演進過程

• 挑戰(zhàn)及實踐

• 總結(jié)與展望

01

相關背景介紹

1. 進口業(yè)務簡介

進口業(yè)務的流程大致比較清晰，國內(nèi)的買家下單之后，國外的賣家發(fā)貨，經(jīng)過清關，干線運輸，到國內(nèi)的清關，配送，到消費者手里，菜鳥在整個過程中負責協(xié)調(diào)鏈路上的各個資源，完成物流履約的服務。去年考拉融入到阿里體系之后，整個進口業(yè)務規(guī)模占國內(nèi)進口單量的規(guī)模是非常高的。并且每年的單量都在迅速增長，訂單履行周期特別長，中間涉及的環(huán)節(jié)多，所以在數(shù)據(jù)建設時，既要考慮把所有數(shù)據(jù)融合到一起，還要保證數(shù)據(jù)有效性，是非常困難的一件事情。

2. 實時數(shù)倉加工流程

① 一般過程

下面簡單介紹一下實時數(shù)倉的加工流程，一般會對接業(yè)務庫或者日志源，通過數(shù)據(jù)同步的方式，比如Sqoop或DataX把消息同步到消息中間件中暫存，下游會接一個實時計算引擎，對消息進行消費，消費之后會進行計算、加工，產(chǎn)出一些明細表或匯總指標，放到查詢服務上供數(shù)據(jù)應用端使用。

② 菜鳥內(nèi)部流程

在菜鳥內(nèi)部也是同樣的流程，我們將業(yè)務庫數(shù)據(jù)通過DRC ( 數(shù)據(jù)備份中心 ) 增量采集Binlog日志的方式，同步到TT ( 類似Kafka的消息中間件 ) 做一個消息暫存，后面會接一個Flink實時計算引擎進行消費，計算好之后寫入兩種查詢服務，一種是ADB，一種是HBase ( Lindorm )，ADB是一個OLAP引擎，阿里云對外也提供服務，主要是提供一些豐富的多維分析查詢，寫入的也是一些維度比較豐富的輕度匯總或明細數(shù)據(jù)，對于實時大屏的場景，因為維度比較少，指標比較固定，我們會沉淀一些高度匯總指標寫到HBase中供實時大屏使用。

02

進口實時數(shù)倉演進過程

接下來講一下進口實時數(shù)倉的演進過程：

2014年：進口業(yè)務線大概在14年時，建好了離線數(shù)倉，能提供日報。

2015年：能提供小時報，更新頻度從天到小時。

2016年：基于JStorm探索了一些實時指標的計算服務，越來越趨向于實時化。由于16年剛開始嘗試實時指標，指標還不是特別豐富。

2017年：菜鳥引進了Blink，也就是Flink在阿里的內(nèi)部版本，作為我們的流計算引擎，并且進口業(yè)務線在同一年打通了實時明細，通過實時明細大寬表對外提供數(shù)據(jù)服務。

2018年：完成了菜鳥進口實時數(shù)倉1.0的建設。

2020年：開始了實時數(shù)倉2.0的建設，為什么開始2.0？因為1.0在設計過程中存在了很多問題，整個模型架構(gòu)不夠靈活，擴展性不高，還有一些是因為沒有了解Blink的特性，導致誤用帶來的一些運維成本的增加，所以后面進行了大的升級改造。

1. 實時數(shù)倉1.0

接下來講一下實時數(shù)倉1.0的情況，一開始因為在發(fā)展初期，業(yè)務模式不太穩(wěn)定，所以一開始的策略就是圍繞業(yè)務小步快跑，比如針對業(yè)務1會開發(fā)一套實時明細層，針對業(yè)務2也會開發(fā)一套實時任務，好處是可以隨著業(yè)務發(fā)展快速迭代，互相之間不影響，早期會更靈活。

如上圖右側(cè)所示，最底層是各個業(yè)務系統(tǒng)的消息源，實時任務主要有兩層，一層是實時明細層，針對業(yè)務線會開發(fā)不同的明細表，明細表就是針對該條業(yè)務線需要的數(shù)據(jù)把它抽取過來，在這之上是ADM層，也就是實時應用層，應用層主要針對具體的場景定制，比如有個場景要看整體匯總指標，則從各個明細表抽取數(shù)據(jù)，產(chǎn)生一張實時匯總層表，整個過程是豎向煙囪式開發(fā)，模型比較混亂，難擴展，并且存在很多重復計算。

后面也是由于重復計算的問題，進行了一層抽象，加了一個前置中間層，對公共的部分進行提取，但是治標不治本，整個模型還是比較混亂的，數(shù)據(jù)建設上也沒有進行統(tǒng)一，模型擴展性上也很差。

2. 實時數(shù)倉2.0

2.0升級完之后是比較清晰的一張圖：

• 前置層：底層數(shù)據(jù)源會接入到前置中間層，屏蔽掉底層一些非常復雜的邏輯。

• 明細層：前置層會把比較干凈的數(shù)據(jù)給到明細表，明細層打通了各個業(yè)務線，進行了模型的統(tǒng)一。

• 匯總層：明細層之上會有輕度匯總和高度匯總，輕度匯總表維度非常多，主要寫入到OLAP引擎中供多維查詢分析，高度匯總指標主要針對實時大屏場景進行沉淀。

• 接口服務：匯總層之上會根據(jù)統(tǒng)一的接口服務對外提供數(shù)據(jù)輸出。

• 數(shù)據(jù)應用：應用層主要接入包括實時大屏，數(shù)據(jù)應用，實時報表以及消息推送等。

這就是實時數(shù)倉2.0升級之后的模型，整個模型雖然看起來比較簡單，其實背后從模型設計到開發(fā)落地，遇到了很多困難，花費了很大的精力。下面為大家分享下我們在升級過程中遇到的挑戰(zhàn)及實踐。

03

挑戰(zhàn)及實踐

我們在實時數(shù)倉升級的過程中，面臨的挑戰(zhàn)如下：

1. 業(yè)務線和業(yè)務模式多

第一個就是對接的業(yè)務線比較多，不同的業(yè)務線有不同的模式，導致一開始小步快跑方式的模型比較割裂，模型和模型之間沒有復用性，開發(fā)和運維成本都很高，資源消耗嚴重。

解決方案：邏輯中間層升級

我們想到的比較簡單的思路就是建設統(tǒng)一的數(shù)據(jù)中間層，比如業(yè)務A有出庫、攬收、派送等幾個業(yè)務節(jié)點，業(yè)務B可能是另外幾個節(jié)點，整個模型是割裂的狀態(tài)，但實際上業(yè)務發(fā)展到中后期比較穩(wěn)定的時候，各個業(yè)務模式之間相對比較穩(wěn)定，這個時候可以對數(shù)據(jù)進行一個抽象，比如業(yè)務A有節(jié)點1、節(jié)點5和其他幾個業(yè)務模式是一樣的，通過這種對齊的方式，找出哪些是公共的，哪些是非公共的，提取出來沉淀到邏輯中間層里，從而屏蔽各業(yè)務之間的差距，完成統(tǒng)一的數(shù)據(jù)建設。把邏輯中間層進行統(tǒng)一，還有一個很大的原因，業(yè)務A，B，C雖然是不同的業(yè)務系統(tǒng)，比如履行系統(tǒng)，關務系統(tǒng)，但是本質(zhì)上都是同一套，底層數(shù)據(jù)源也是進行各種抽象，所以數(shù)倉建模上也要通過統(tǒng)一的思路進行建設。

2. 業(yè)務系統(tǒng)多，超大數(shù)據(jù)源

第二個就是對接的系統(tǒng)非常多，每個系統(tǒng)數(shù)據(jù)量很大，每天億級別的數(shù)據(jù)源就有十幾個，梳理起來非常困難。帶來的問題也比較明顯，第一個問題就是大狀態(tài)的問題，需要在Flink里維護特別大的狀態(tài)，還有就是接入這么多數(shù)據(jù)源之后，成本怎么控制。

解決方案：善用State

State是Flink的一大特性，因為它才能保證狀態(tài)計算，需要更合理的利用。我們要認清State是干什么的，什么時候需要State，如何優(yōu)化它，這些都是需要考慮的事情。State有兩種，一種是KeyedState，具體是跟數(shù)據(jù)的Key相關的，例如SQL中的Group By，F(xiàn)link會按照鍵值進行相關數(shù)據(jù)的存儲，比如存儲到二進制的一個數(shù)組里。第二個是OperatorState，跟具體的算子相關，比如用來記錄Source Connector里讀取的Offset，或者算子之間任務Failover之后，狀態(tài)怎么在不同算子之間進行恢復。

① 數(shù)據(jù)接入時"去重"

下面舉個例子，怎么用到KeyedState，比如物流訂單流和履行日志流，兩個作業(yè)關聯(lián)產(chǎn)生出最終需要的一張大表，Join是怎么存儲的呢？流是一直不停的過來的，消息到達的前后順序可能不一致，需要把它存在算子里面，對于Join的狀態(tài)節(jié)點，比較簡單粗暴的方式是把左流和右流同時存下來，通過這樣的方式保證不管消息是先到還是后到，至少保證算子里面數(shù)據(jù)是全的，哪怕其中一個流很晚才到達，也能保證匹配到之前的數(shù)據(jù)，需要注意的一點是，State存儲根據(jù)上游不同而不同，比如在上游定義了一個主鍵Rowkey，并且JoinKey包含了主鍵，就不存在多筆訂單對應同一個外鍵，這樣就告訴State只需要按照JoinKey存儲唯一行就可以了。如果上游有主鍵，但是JoinKey不包含Rowkey 的話，就需要在State里將兩個Rowkey的訂單同時存下來。最差的情況是，上游沒有主鍵，比如同一筆訂單有10條消息，會有先后順序，最后一條是有效的，但是對于系統(tǒng)來說不知道哪條是有效的，沒有指定主鍵也不好去重，它就會全部存下來，特別耗資源和性能，相對來說是特別差的一種方式。

因此，我們在數(shù)據(jù)接入時進行"去重"。數(shù)據(jù)接入時，按照row_number進行排序，告訴系統(tǒng)按照主鍵進行數(shù)據(jù)更新就可以了，解決10條消息不知道應該存幾條的問題。在上面這個case里面，就是按照主鍵進行更新，每次取最后一條消息。

按照row_number這種方式并不會減少數(shù)據(jù)處理量，但是會大大減少State存儲量，每一個State只存一份有效的狀態(tài)，而不是把它所有的歷史數(shù)據(jù)都記錄下來。

② 多流join優(yōu)化

第二個是多流Join的優(yōu)化，比如像上圖左側(cè)的偽代碼，一張主表關聯(lián)很多數(shù)據(jù)源產(chǎn)生一個明細大寬表，這是我們喜歡的方式，但是這樣并不好，為什么呢？這樣一個SQL在實時計算里會按照雙流Join的方式依次處理，每次只能處理一個Join，所以像左邊這個代碼里有10個Join，在右邊就會有10個Join節(jié)點，Join節(jié)點會同時將左流和右流的數(shù)據(jù)全部存下來，所以會看到右邊這個圖的紅框里，每一個Join節(jié)點會同時存儲左流和右流的節(jié)點，假設我們訂單源有1億，里面存的就是10億，這個數(shù)據(jù)量存儲是非?？膳碌摹?/span>

另外一個就是鏈路特別長，不停的要進行網(wǎng)絡傳輸，計算，任務延遲也是很大的。像十幾個數(shù)據(jù)源取數(shù)關聯(lián)在一起，在我們的實際場景是真實存在的，而且我們的關聯(lián)關系比這個還要更復雜。

那我們怎么優(yōu)化呢？我們采用Union All的方式，把數(shù)據(jù)錯位拼接到一起，后面加一層Group By，相當于將Join關聯(lián)轉(zhuǎn)換成Group By，它的執(zhí)行圖就像上圖右側(cè)這樣，黃色是數(shù)據(jù)接入過程中需要進行的存儲，紅色是一個Join節(jié)點，所以整個過程需要存儲的State是非常少的，主表會在黃色框和紅色框分別存一份，別看數(shù)據(jù)源非常多，其實只會存一份數(shù)據(jù)，比如我們的物流訂單是1000萬，其他數(shù)據(jù)源也是1000萬，最終的結(jié)果有效行就是1000萬，數(shù)據(jù)存儲量其實是不高的，假設又新接了數(shù)據(jù)源，可能又是1000萬的日志量，但其實有效記錄就是1000萬，只是增加了一個數(shù)據(jù)源，進行了一個數(shù)據(jù)更新，新增數(shù)據(jù)源成本近乎為0，所以用Union All替換Join的方式在State里是一個大大的優(yōu)化。

3. 取數(shù)外鍵多，易亂序

第三個是取數(shù)外鍵多，亂序的問題，亂序其實有很多種，采集系統(tǒng)采集過來就是亂序的，或者傳輸過程中導致的亂序，我們這邊要討論的是，在實際開發(fā)過程中不小心導致的亂序，因為其他層面的東西平臺已經(jīng)幫我們考慮好了，提供了很好的端到端的一致性保證。

舉個例子比如說有兩個單子都是物流單，根據(jù)單號取一些倉內(nèi)的消息，消息1和消息2先后進入流處理里面，關聯(lián)的時候根據(jù)JoinKey進行Shuffle，在這種情況下，兩個消息會流到不同的算子并發(fā)上，如果這兩個并發(fā)處理速度不一致，就有可能導致先進入系統(tǒng)的消息后完成處理，比如消息1先到達系統(tǒng)的，但是處理比較慢，消息2反倒先產(chǎn)出，導致最終的輸出結(jié)果是不對的，本質(zhì)上是多并發(fā)場景下，數(shù)據(jù)處理流向的不確定性，同一筆訂單的多筆消息流到不同的地方進行計算，就可能會導致亂序。

所以，同一筆訂單消息處理完之后，如何保證是有序的？

上圖是一個簡化的過程，業(yè)務庫流入到Kafka，Binlog日志是順序?qū)懭氲?，需要采用一定的策略，也是順序采集，可以根?jù)主鍵進行Hash分區(qū)，寫到Kafka里面，保證Kafka里面每個分區(qū)存的數(shù)據(jù)是同一個Key，首先在這個層面保證有序。然后Flink消費Kafka時，需要設置合理的并發(fā)，保證一個分區(qū)的數(shù)據(jù)由一個Operator負責，如果一個分區(qū)由兩個Operator負責，就會存在類似于剛才的情況，導致消息亂序。另外還要配合下游的應用，能保證按照某些主鍵進行更新或刪除操作，這樣才能保證端到端的一致性。

Flink已經(jīng)配合上下游系統(tǒng)已經(jīng)幫我們實現(xiàn)了端到端的一致性功能，我們只需要保證內(nèi)部處理任務不能亂序。我們的解法是避免Join Key發(fā)生變化，如提前通過特殊映射關系把Join Key變?yōu)闃I(yè)務主鍵，來保證任務處理是有序的。

4. 統(tǒng)計指標依賴明細，服務壓力大

另外一個難點就是我們的很多統(tǒng)計指標都依賴明細，主要是一些實時統(tǒng)計，這種風險比較明顯，服務端壓力特別大，尤其是大促時，極其容易把系統(tǒng)拖垮。

實時超時統(tǒng)計就是一個典型的場景，比如說會有這樣兩筆訂單，一筆訂單1點鐘創(chuàng)建了物流訂單，2點鐘進行出庫，如何統(tǒng)計超6小時未攬收的收單量，因為沒有消息就無法觸發(fā)計算，F(xiàn)link是基于消息觸發(fā)的，比如說2點鐘出庫了，那理論上在8點鐘的時候超6小時未攬收的單量要加1，但是因為沒有消息觸發(fā)，下游系統(tǒng)不會觸發(fā)計算，這是比較難的事情，所以一開始沒有特別好的方案，我們直接從明細表出，比如訂單的出庫時間是2點鐘，生成這條明細之后，寫到數(shù)據(jù)庫的OLAP引擎里，和當前明細進行比較計算。

我們也探索了一些方案比如基于消息中間件，進行一些定時超時消息下發(fā)，或者也探索過基于Flink CEP的方式，第一種方式需要引入第三方的中間件，維護成本會更高，CEP這種方式采用時間窗口穩(wěn)步向前走，像我們這種物流場景下會存在很多這樣的情況，比如回傳一個2點出庫的時間，后面發(fā)現(xiàn)回傳錯了，又會補一個1點半的時間，那么我們需要重新觸發(fā)計算，F(xiàn)link CEP是不能很好的支持的。后面我們探索了基于Flink Timer Service這種方式，基于Flink自帶的Timer Service回調(diào)方法，來制造一個消息流，首先在我們的方法里面接入數(shù)據(jù)流，根據(jù)我們定義的一些規(guī)則，比如出庫時間是2點，會定義6小時的一個超時時間，注冊到Timer Service里面，到8點會觸發(fā)一次比較計算，沒有的話就會觸發(fā)一個超時消息，整個方案不依賴第三方組件，開發(fā)成本比較低。

5. 履行環(huán)節(jié)多，數(shù)據(jù)鏈路長

另外一個難點就是我們的履行環(huán)節(jié)比較多，數(shù)據(jù)鏈路比較長，導致異常情況很難處理。比如消息要保留20多天的有效期，State也要存20多天，狀態(tài)一直存在Flink里面，如果某一天數(shù)據(jù)出現(xiàn)錯誤或者邏輯加工錯誤，追溯是個很大問題，因為上游的消息系統(tǒng)一般保持三天數(shù)據(jù)的有效期。

這邊說幾個真實的案例。

案例1：

我們在雙十一期間發(fā)現(xiàn)了一個Bug，雙十一已經(jīng)過去好幾天了，因為我們的履行鏈路特別長，要10~20天，第一時間發(fā)現(xiàn)錯誤要改已經(jīng)改不了了，改了之后DAG執(zhí)行圖會發(fā)生變化，狀態(tài)就無法恢復，而且上游只能追3天的數(shù)，改了之后相當于上游的數(shù)全沒了，這是不能接受的。

案例2：

疫情期間的一些超長尾單，State的TTL設置都是60天，我們認為60天左右肯定能夠全部完結(jié)，后來發(fā)現(xiàn)超過24天數(shù)據(jù)開始失真，明明設置的有效期是60天，后來發(fā)現(xiàn)底層State存儲用的是int型，所以最多只能存20多天的有效期，相當于觸發(fā)了Flink的一個邊界case，所以也證明了我們這邊的場景的確很復雜，很多狀態(tài)需要超長的State生命周期來保證的。

案例3：

每次代碼停止升級之后，狀態(tài)就丟失了，需要重新拉取數(shù)據(jù)計算，但是一般上游的數(shù)據(jù)只保留3天有效期，這樣的話業(yè)務只能看3天的數(shù)據(jù)，用戶體驗很不好。

解決方案：批流混合

我們怎么做？

采用批流混合的方式來完成狀態(tài)復用，基于Blink流處理來處理實時消息流，基于Blink的批處理完成離線計算，通過兩者的融合，在同一個任務里完成歷史所有數(shù)據(jù)的計算，舉個例子，訂單消息流和履行消息流進行一個關聯(lián)計算，那么會在任務里增加一個離線訂單消息源，跟我們的實時訂單消息源Union All合并在一起，下面再增加一個Group By節(jié)點，按照主鍵進行去重，基于這種方式就可以實現(xiàn)狀態(tài)復用。有幾個需要注意的點，第一個需要自定義Source Connector去開發(fā)，另外一個涉及到離線消息和實時消息合并的一個問題，GroupBy之后是優(yōu)先取離線消息還是實時消息，實時消息可能消費的比較慢，哪個消息是真實有效的需要判斷一下，所以我們也定制了一些，比如LastValue來解決任務是優(yōu)先取離線消息還是實時消息，整個過程是基于Blink和MaxCompute來實現(xiàn)的。

6. 一些小的Tips

① 消息下發(fā)無法撤回問題

第一個就是消息一旦下發(fā)無法撤回，所以有些訂單一開始有效，后面變成無效了，這種訂單不應該在任務中過濾，而是打上標記下傳，統(tǒng)計的時候再用。

② 增加數(shù)據(jù)版本，數(shù)據(jù)處理時間以及數(shù)據(jù)處理版本

• 數(shù)據(jù)版本是消息結(jié)構(gòu)體的版本定義，避免模型升級后，任務重啟讀到臟數(shù)據(jù)。

• 處理時間就是消息當前的處理時間，比如消息回流到離線，我們會按照主鍵進行時間排序，取到最新記錄，通過這種方式還原一份準實時數(shù)據(jù)。

• 增加數(shù)據(jù)處理版本是因為即使到毫秒級也不夠精確，無法區(qū)分消息的前后順序。

③ 實時對數(shù)方案

實時對數(shù)方案有兩個層面，實時明細和離線明細，剛剛也提到將實時數(shù)據(jù)回流到離線，我們可以看當前24點前產(chǎn)生的消息，因為離線T+1只能看到昨天23點59分59秒的數(shù)據(jù)，實時也可以模擬，我們只截取那個時刻的數(shù)據(jù)還原出來，然后實時和離線進行對比，這樣也可以很好的進行數(shù)據(jù)比對，另外可以進行實時明細和實時匯總對比，因為都在同一個DB里，對比起來也特別方便。

03

總結(jié)與展望

1. 總結(jié)

簡單做下總結(jié)：

• 模型與架構(gòu)：好的模型和架構(gòu)相當于成功了80%。

• 準確性要求評估：需要評估數(shù)據(jù)準確性要求，是否真的需要對齊CheckPoint或者一致性的語義保證，有些情況下保證一般準確性就ok了，那么就不需要這么多額外消耗資源的設計。

• 合理利用Flink特性：需要合理利用Fink的一些特性，避免一些誤用之痛，比如State和CheckPoint的使用。

• 代碼自查：保證數(shù)據(jù)處理是正常流轉(zhuǎn)的，合乎目標。

• SQL理解：寫SQL并不是有多高大上，更多考驗的是在數(shù)據(jù)流轉(zhuǎn)過程中的一些思考。

2. 展望

① 實時數(shù)據(jù)質(zhì)量監(jiān)控

實時處理不像批處理，批處理跑完之后可以在跑個小腳本統(tǒng)計一下主鍵是否唯一，記錄數(shù)波動等，實時的數(shù)據(jù)監(jiān)控是比較麻煩的事情·。

② 流批統(tǒng)一

流批統(tǒng)一有幾個層面，第一個就是存儲層面的統(tǒng)一，實時和離線寫到同一個地方去，應用的時候更方便。第二個就是計算引擎的統(tǒng)一，比如像Flink可以同時支持批處理和流處理，還能夠?qū)懙紿ive里面。更高層次的就是可以做到處理結(jié)果的統(tǒng)一，同一段代碼，在批和流的語義可能會不一樣，如何做到同一段代碼，批和流的處理結(jié)果是完全統(tǒng)一的。

③ 自動調(diào)優(yōu)

自動調(diào)優(yōu)有兩種，比如在大促的時候，我們申請了1000個Core的資源，1000個Core怎么合理的分配，哪些地方可能是性能瓶頸，要多分配一些，這是給定資源的自動調(diào)優(yōu)。還有一種比如像凌晨沒什么單量，也沒什么數(shù)據(jù)流量，這個時候可以把資源調(diào)到很小，根據(jù)數(shù)據(jù)流量情況自動調(diào)整，也就是自動伸縮能力。

以上是我們整體對未來的展望和研究方向。??????????????????????

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

長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝