簡介：供應鏈物流場景下的業務複雜度高，業務鏈路長，節點多，實體多，實時數倉建設難度高。菜鳥跨境進口業務場景更是如此，更複雜的場景帶來更複雜的實體數據模型，對接的業務系統多導致ETL流程特別複雜，還有海量的日均處理數據量，使得團隊在建設進口實時數倉的過程中，面臨著諸多挑戰。
如何保證複雜實體關係下的數據準確性？如何降低多數據源情況下的數據處理複雜度？如何提升實時多流Join的處理效率？如何實現實時超時統計？如何實現異常情況下的數據狀態恢復？本文主要分享菜鳥進口實時數倉的升級經驗，以及如何利用Flink的特性解決在開發實踐中遇到的問題。
主要內容包括：
相關背景介紹進口實時數倉演進過程挑戰及實踐總結與展望01 相關背景介紹
1. 進口業務簡介
進口業務的流程大致比較清晰，國內的買家下單之後，國外的賣家發貨，經過清關，幹線運輸，到國內的清關，配送，到消費者手裡，菜鳥在整個過程中負責協調鏈路上的各個資源，完成物流履約的服務。去年考拉融入到阿里體系之後，整個進口業務規模佔國內進口單量的規模是非常高的。並且每年的單量都在迅速增長，訂單履行周期特別長，中間涉及的環節多，所以在數據建設時，既要考慮把所有數據融合到一起，還要保證數據有效性，是非常困難的一件事情。
2. 實時數倉加工流程
① 一般過程
下面簡單介紹一下實時數倉的加工流程，一般會對接業務庫或者日誌源，通過數據同步的方式，比如Sqoop或DataX把消息同步到消息中間件中暫存，下遊會接一個實時計算引擎，對消息進行消費，消費之後會進行計算、加工，產出一些明細表或匯總指標，放到查詢服務上供數據應用端使用。
② 菜鳥內部流程
在菜鳥內部也是同樣的流程，我們將業務庫數據通過DRC ( 數據備份中心 ) 增量採集Binlog日誌的方式，同步到TT ( 類似Kafka的消息中間件 ) 做一個消息暫存，後面會接一個Flink實時計算引擎進行消費，計算好之後寫入兩種查詢服務，一種是ADB，一種是HBase ( Lindorm )，ADB是一個OLAP引擎，阿里雲對外也提供服務，主要是提供一些豐富的多維分析查詢，寫入的也是一些維度比較豐富的輕度匯總或明細數據，對於實時大屏的場景，因為維度比較少，指標比較固定，我們會沉澱一些高度匯總指標寫到HBase中供實時大屏使用。
02 進口實時數倉演進過程
接下來講一下進口實時數倉的演進過程：
2014年：進口業務線大概在14年時，建好了離線數倉，能提供日報。
2015年：能提供小時報，更新頻度從天到小時。
2016年：基於JStorm探索了一些實時指標的計算服務，越來越趨向於實時化。由於16年剛開始嘗試實時指標，指標還不是特別豐富。
2017年：菜鳥引進了Blink，也就是Flink在阿里的內部版本，作為我們的流計算引擎，並且進口業務線在同一年打通了實時明細，通過實時明細大寬表對外提供數據服務。
2018年：完成了菜鳥進口實時數倉1.0的建設。
2020年：開始了實時數倉2.0的建設，為什麼開始2.0？因為1.0在設計過程中存在了很多問題，整個模型架構不夠靈活，擴展性不高，還有一些是因為沒有了解Blink的特性，導致誤用帶來的一些運維成本的增加，所以後面進行了大的升級改造。
1. 實時數倉1.0
接下來講一下實時數倉1.0的情況，一開始因為在發展初期，業務模式不太穩定，所以一開始的策略就是圍繞業務小步快跑，比如針對業務1會開發一套實時明細層，針對業務2也會開發一套實時任務，好處是可以隨著業務發展快速迭代，互相之間不影響，早期會更靈活。
如上圖右側所示，最底層是各個業務系統的消息源，實時任務主要有兩層，一層是實時明細層，針對業務線會開發不同的明細表，明細表就是針對該條業務線需要的數據把它抽取過來，在這之上是ADM層，也就是實時應用層，應用層主要針對具體的場景定製，比如有個場景要看整體匯總指標，則從各個明細表抽取數據，產生一張實時匯總層表，整個過程是豎向煙囪式開發，模型比較混亂，難擴展，並且存在很多重複計算。
後面也是由於重複計算的問題，進行了一層抽象，加了一個前置中間層，對公共的部分進行提取，但是治標不治本，整個模型還是比較混亂的，數據建設上也沒有進行統一，模型擴展性上也很差。
2. 實時數倉2.0
2.0升級完之後是比較清晰的一張圖：
前置層：底層數據源會接入到前置中間層，屏蔽掉底層一些非常複雜的邏輯。
明細層：前置層會把比較乾淨的數據給到明細表，明細層打通了各個業務線，進行了模型的統一。匯總層：明細層之上會有輕度匯總和高度匯總，輕度匯總表維度非常多，主要寫入到OLAP引擎中供多維查詢分析，高度匯總指標主要針對實時大屏場景進行沉澱。接口服務：匯總層之上會根據統一的接口服務對外提供數據輸出。數據應用：應用層主要接入包括實時大屏，數據應用，實時報表以及消息推送等。這就是實時數倉2.0升級之後的模型，整個模型雖然看起來比較簡單，其實背後從模型設計到開發落地，遇到了很多困難，花費了很大的精力。下面為大家分享下我們在升級過程中遇到的挑戰及實踐。
03 挑戰及實踐
我們在實時數倉升級的過程中，面臨的挑戰如下：
1. 業務線和業務模式多
第一個就是對接的業務線比較多，不同的業務線有不同的模式，導致一開始小步快跑方式的模型比較割裂，模型和模型之間沒有復用性，開發和運維成本都很高，資源消耗嚴重。
解決方案：邏輯中間層升級
我們想到的比較簡單的思路就是建設統一的數據中間層，比如業務A有出庫、攬收、派送等幾個業務節點，業務B可能是另外幾個節點，整個模型是割裂的狀態，但實際上業務發展到中後期比較穩定的時候，各個業務模式之間相對比較穩定，這個時候可以對數據進行一個抽象，比如業務A有節點1、節點5和其他幾個業務模式是一樣的，通過這種對齊的方式，找出哪些是公共的，哪些是非公共的，提取出來沉澱到邏輯中間層裡，從而屏蔽各業務之間的差距，完成統一的數據建設。把邏輯中間層進行統一，還有一個很大的原因，業務A，B，C雖然是不同的業務系統，比如履行系統，關務系統，但是本質上都是同一套，底層數據源也是進行各種抽象，所以數倉建模上也要通過統一的思路進行建設。
2. 業務系統多，超大數據源
第二個就是對接的系統非常多，每個系統數據量很大，每天億級別的數據源就有十幾個，梳理起來非常困難。帶來的問題也比較明顯，第一個問題就是大狀態的問題，需要在Flink裡維護特別大的狀態，還有就是接入這麼多數據源之後，成本怎麼控制。
解決方案：善用State
State是Flink的一大特性，因為它才能保證狀態計算，需要更合理的利用。我們要認清State是幹什麼的，什麼時候需要State，如何優化它，這些都是需要考慮的事情。State有兩種，一種是KeyedState，具體是跟數據的Key相關的，例如SQL中的Group By，Flink會按照鍵值進行相關數據的存儲，比如存儲到二進位的一個數組裡。第二個是OperatorState，跟具體的算子相關，比如用來記錄Source Connector裡讀取的Offset，或者算子之間任務Failover之後，狀態怎麼在不同算子之間進行恢復。
① 數據接入時"去重"
下面舉個例子，怎麼用到KeyedState，比如物流訂單流和履行日誌流，兩個作業關聯產生出最終需要的一張大表，Join是怎麼存儲的呢？流是一直不停的過來的，消息到達的前後順序可能不一致，需要把它存在算子裡面，對於Join的狀態節點，比較簡單粗暴的方式是把左流和右流同時存下來，通過這樣的方式保證不管消息是先到還是後到，至少保證算子裡面數據是全的，哪怕其中一個流很晚才到達，也能保證匹配到之前的數據，需要注意的一點是，State存儲根據上遊不同而不同，比如在上遊定義了一個主鍵Rowkey，並且JoinKey包含了主鍵，就不存在多筆訂單對應同一個外鍵，這樣就告訴State只需要按照JoinKey存儲唯一行就可以了。如果上遊有主鍵，但是JoinKey不包含Rowkey 的話，就需要在State裡將兩個Rowkey的訂單同時存下來。最差的情況是，上遊沒有主鍵，比如同一筆訂單有10條消息，會有先後順序，最後一條是有效的，但是對於系統來說不知道哪條是有效的，沒有指定主鍵也不好去重，它就會全部存下來，特別耗資源和性能，相對來說是特別差的一種方式。
因此，我們在數據接入時進行"去重"。數據接入時，按照row_number進行排序，告訴系統按照主鍵進行數據更新就可以了，解決10條消息不知道應該存幾條的問題。在上面這個case裡面，就是按照主鍵進行更新，每次取最後一條消息。
按照row_number這種方式並不會減少數據處理量，但是會大大減少State存儲量，每一個State只存一份有效的狀態，而不是把它所有的歷史數據都記錄下來。
② 多流join優化
第二個是多流Join的優化，比如像上圖左側的偽代碼，一張主表關聯很多數據源產生一個明細大寬表，這是我們喜歡的方式，但是這樣並不好，為什麼呢？這樣一個SQL在實時計算裡會按照雙流Join的方式依次處理，每次只能處理一個Join，所以像左邊這個代碼裡有10個Join，在右邊就會有10個Join節點，Join節點會同時將左流和右流的數據全部存下來，所以會看到右邊這個圖的紅框裡，每一個Join節點會同時存儲左流和右流的節點，假設我們訂單源有1億，裡面存的就是10億，這個數據量存儲是非常可怕的。
另外一個就是鏈路特別長，不停的要進行網絡傳輸，計算，任務延遲也是很大的。像十幾個數據源取數關聯在一起，在我們的實際場景是真實存在的，而且我們的關聯關係比這個還要更複雜。
那我們怎麼優化呢？我們採用Union All的方式，把數據錯位拼接到一起，後面加一層Group By，相當於將Join關聯轉換成Group By，它的執行圖就像上圖右側這樣，黃色是數據接入過程中需要進行的存儲，紅色是一個Join節點，所以整個過程需要存儲的State是非常少的，主表會在黃色框和紅色框分別存一份，別看數據源非常多，其實只會存一份數據，比如我們的物流訂單是1000萬，其他數據源也是1000萬，最終的結果有效行就是1000萬，數據存儲量其實是不高的，假設又新接了數據源，可能又是1000萬的日誌量，但其實有效記錄就是1000萬，只是增加了一個數據源，進行了一個數據更新，新增數據源成本近乎為0，所以用Union All替換Join的方式在State裡是一個大大的優化。
3. 取數外鍵多，易亂序
第三個是取數外鍵多，亂序的問題，亂序其實有很多種，採集系統採集過來就是亂序的，或者傳輸過程中導致的亂序，我們這邊要討論的是，在實際開發過程中不小心導致的亂序，因為其他層面的東西平臺已經幫我們考慮好了，提供了很好的端到端的一致性保證。
舉個例子比如說有兩個單子都是物流單，根據單號取一些倉內的消息，消息1和消息2先後進入流處理裡面，關聯的時候根據JoinKey進行Shuffle，在這種情況下，兩個消息會流到不同的算子並發上，如果這兩個並發處理速度不一致，就有可能導致先進入系統的消息後完成處理，比如消息1先到達系統的，但是處理比較慢，消息2反倒先產出，導致最終的輸出結果是不對的，本質上是多並發場景下，數據處理流向的不確定性，同一筆訂單的多筆消息流到不同的地方進行計算，就可能會導致亂序。
所以，同一筆訂單消息處理完之後，如何保證是有序的？
上圖是一個簡化的過程，業務庫流入到Kafka，Binlog日誌是順序寫入的，需要採用一定的策略，也是順序採集，可以根據主鍵進行Hash分區，寫到Kafka裡面，保證Kafka裡面每個分區存的數據是同一個Key，首先在這個層面保證有序。然後Flink消費Kafka時，需要設置合理的並發，保證一個分區的數據由一個Operator負責，如果一個分區由兩個Operator負責，就會存在類似於剛才的情況，導致消息亂序。另外還要配合下遊的應用，能保證按照某些主鍵進行更新或刪除操作，這樣才能保證端到端的一致性。
Flink已經配合上下遊系統已經幫我們實現了端到端的一致性功能，我們只需要保證內部處理任務不能亂序。我們的解法是避免Join Key發生變化，如提前通過特殊映射關係把Join Key變為業務主鍵，來保證任務處理是有序的。
4. 統計指標依賴明細，服務壓力大
另外一個難點就是我們的很多統計指標都依賴明細，主要是一些實時統計，這種風險比較明顯，服務端壓力特別大，尤其是大促時，極其容易把系統拖垮。
實時超時統計就是一個典型的場景，比如說會有這樣兩筆訂單，一筆訂單1點鐘創建了物流訂單，2點鐘進行出庫，如何統計超6小時未攬收的收單量，因為沒有消息就無法觸發計算，Flink是基於消息觸發的，比如說2點鐘出庫了，那理論上在8點鐘的時候超6小時未攬收的單量要加1，但是因為沒有消息觸發，下遊系統不會觸發計算，這是比較難的事情，所以一開始沒有特別好的方案，我們直接從明細表出，比如訂單的出庫時間是2點鐘，生成這條明細之後，寫到資料庫的OLAP引擎裡，和當前明細進行比較計算。
我們也探索了一些方案比如基於消息中間件，進行一些定時超時消息下發，或者也探索過基於Flink CEP的方式，第一種方式需要引入第三方的中間件，維護成本會更高，CEP這種方式採用時間窗口穩步向前走，像我們這種物流場景下會存在很多這樣的情況，比如回傳一個2點出庫的時間，後面發現回傳錯了，又會補一個1點半的時間，那麼我們需要重新觸發計算，Flink CEP是不能很好的支持的。後面我們探索了基於Flink Timer Service這種方式，基於Flink自帶的Timer Service回調方法，來製造一個消息流，首先在我們的方法裡面接入數據流，根據我們定義的一些規則，比如出庫時間是2點，會定義6小時的一個超時時間，註冊到Timer Service裡面，到8點會觸發一次比較計算，沒有的話就會觸發一個超時消息，整個方案不依賴第三方組件，開發成本比較低。
5. 履行環節多，數據鏈路長
另外一個難點就是我們的履行環節比較多，數據鏈路比較長，導致異常情況很難處理。比如消息要保留20多天的有效期，State也要存20多天，狀態一直存在Flink裡面，如果某一天數據出現錯誤或者邏輯加工錯誤，追溯是個很大問題，因為上遊的消息系統一般保持三天數據的有效期。
這邊說幾個真實的案例。
案例1：
我們在雙十一期間發現了一個Bug，雙十一已經過去好幾天了，因為我們的履行鏈路特別長，要10~20天，第一時間發現錯誤要改已經改不了了，改了之後DAG執行圖會發生變化，狀態就無法恢復，而且上遊只能追3天的數，改了之後相當於上遊的數全沒了，這是不能接受的。
案例2：
疫情期間的一些超長尾單，State的TTL設置都是60天，我們認為60天左右肯定能夠全部完結，後來發現超過24天數據開始失真，明明設置的有效期是60天，後來發現底層State存儲用的是int型，所以最多只能存20多天的有效期，相當於觸發了Flink的一個邊界case，所以也證明了我們這邊的場景的確很複雜，很多狀態需要超長的State生命周期來保證的。
案例3：
每次代碼停止升級之後，狀態就丟失了，需要重新拉取數據計算，但是一般上遊的數據只保留3天有效期，這樣的話業務只能看3天的數據，用戶體驗很不好。
解決方案：批流混合
我們怎麼做？
用批流混合的方式來完成狀態復用，基於Blink流處理來處理實時消息流，基於Blink的批處理完成離線計算，通過兩者的融合，在同一個任務裡完成歷史所有數據的計算，舉個例子，訂單消息流和履行消息流進行一個關聯計算，那麼會在任務裡增加一個離線訂單消息源，跟我們的實時訂單消息源Union All合併在一起，下面再增加一個Group By節點，按照主鍵進行去重，基於這種方式就可以實現狀態復用。有幾個需要注意的點，第一個需要自定義Source Connector去開發，另外一個涉及到離線消息和實時消息合併的一個問題，GroupBy之後是優先取離線消息還是實時消息，實時消息可能消費的比較慢，哪個消息是真實有效的需要判斷一下，所以我們也定製了一些，比如LastValue來解決任務是優先取離線消息還是實時消息，整個過程是基於Blink和MaxCompute來實現的。
6. 一些小的Tips
① 消息下發無法撤回問題
第一個就是消息一旦下發無法撤回，所以有些訂單一開始有效，後面變成無效了，這種訂單不應該在任務中過濾，而是打上標記下傳，統計的時候再用。
② 增加數據版本，數據處理時間以及數據處理版本
數據版本是消息結構體的版本定義，避免模型升級後，任務重啟讀到髒數據。處理時間就是消息當前的處理時間，比如消息回流到離線，我們會按照主鍵進行時間排序，取到最新記錄，通過這種方式還原一份準實時數據。增加數據處理版本是因為即使到毫秒級也不夠精確，無法區分消息的前後順序。③ 實時對數方案
實時對數方案有兩個層面，實時明細和離線明細，剛剛也提到將實時數據回流到離線，我們可以看當前24點前產生的消息，因為離線T+1隻能看到昨天23點59分59秒的數據，實時也可以模擬，我們只截取那個時刻的數據還原出來，然後實時和離線進行對比，這樣也可以很好的進行數據比對，另外可以進行實時明細和實時匯總對比，因為都在同一個DB裡，對比起來也特別方便。
03 總結與展望
1. 總結
簡單做下總結：
模型與架構：好的模型和架構相當於成功了80%。準確性要求評估：需要評估數據準確性要求，是否真的需要對齊CheckPoint或者一致性的語義保證，有些情況下保證一般準確性就ok了，那麼就不需要這麼多額外消耗資源的設計。合理利用Flink特性：需要合理利用Fink的一些特性，避免一些誤用之痛，比如State和CheckPoint的使用。代碼自查：保證數據處理是正常流轉的，合乎目標。SQL理解：寫SQL並不是有多高大上，更多考驗的是在數據流轉過程中的一些思考。2. 展望
① 實時數據質量監控
實時處理不像批處理，批處理跑完之後可以在跑個小腳本統計一下主鍵是否唯一，記錄數波動等，實時的數據監控是比較麻煩的事情·。
② 流批統一
流批統一有幾個層面，第一個就是存儲層面的統一，實時和離線寫到同一個地方去，應用的時候更方便。第二個就是計算引擎的統一，比如像Flink可以同時支持批處理和流處理，還能夠寫到Hive裡面。更高層次的就是可以做到處理結果的統一，同一段代碼，在批和流的語義可能會不一樣，如何做到同一段代碼，批和流的處理結果是完全統一的。
③ 自動調優
自動調優有兩種，比如在大促的時候，我們申請了1000個Core的資源，1000個Core怎麼合理的分配，哪些地方可能是性能瓶頸，要多分配一些，這是給定資源的自動調優。還有一種比如像凌晨沒什麼單量，也沒什麼數據流量，這個時候可以把資源調到很小，根據數據流量情況自動調整，也就是自動伸縮能力。
以上是我們整體對未來的展望和研究方向。
作者：張庭（菜鳥數據工程師）