1、背景

得物供應鏈業(yè)務是紛繁復雜的，我們既有JIT的現(xiàn)貨模式中間夾著這大量的倉庫作業(yè)環(huán)節(jié)，又有到倉的寄售，品牌業(yè)務，有非常復雜的逆向鏈路。在這么復雜的業(yè)務背后，我們需要精細化關注人貨場車的效率和成本，每一單的及時履約情況，要做到這一點我們需要各粒度和維度的數(shù)據(jù)來支撐我們的精細化管理。

1.1 業(yè)務早期

業(yè)務早期，業(yè)務反饋我們后臺管理系統(tǒng)某些報表查詢慢。查詢代碼可知，如下圖：

這種現(xiàn)象一般表現(xiàn)為：

• 大表JOIN,rdbms不擅長做數(shù)據(jù)聚合,查詢響應慢,調(diào)優(yōu)困難；
• 多表關聯(lián),索引優(yōu)化,子查詢優(yōu)化,加劇了復雜度,大量索引,讀庫磁盤空間膨脹過快；
• 數(shù)據(jù)量大,多維分析困難,跨域取數(shù),自助拉到實時數(shù)據(jù)困難等。

一方面原因是系統(tǒng)設計之初,我們主要關注業(yè)務流程功能設計,事務型業(yè)務流程數(shù)據(jù)建模,對于未來核心指標的落地,特別是關鍵實時指標落地在業(yè)務快速增長的情況下如何做到非常好的支撐。mysql在此方面越來越捉襟見肘。

另外一方面原因是mysql這種oltp數(shù)據(jù)庫是無法滿足實時數(shù)據(jù)分析需求的,我們需要探索一套實時數(shù)據(jù)架構,拉通我們的履約,倉儲,運配等各域的數(shù)據(jù),做有效串聯(lián),因此我們開始了我們的實時數(shù)據(jù)架構探索，下圖是我們一些思考。

附:數(shù)據(jù)視角的架構設計也是系統(tǒng)架構設計的重要組成部分。

2、架構演變

2.1 原始階段

2.1.1 通過Adb(AnalyticDB for MySQL)完成實時join

通過阿里云DTS同步直接將業(yè)務庫單表實時同步到Adb，通過Adb強大的join能力和完全兼容mysql語法,可以執(zhí)行任意sql，對于單表大數(shù)據(jù)量場景或者單表和一些簡單維表的join場景表現(xiàn)還是不錯的,但是在業(yè)務復雜,復雜的sql rt很難滿足要求,即使rt滿足要求,單個sql所消耗的內(nèi)存,cpu也不盡人意,能支撐的并發(fā)量很有限。

2.1.2 通過Otter完成大寬表的建設

基于Canal開源產(chǎn)品，獲取數(shù)據(jù)庫增量日志數(shù)據(jù)并下發(fā)，下游消費增量數(shù)據(jù)直接生成大寬表，但是寬表還是寫入mysql數(shù)據(jù)庫，實現(xiàn)單表查詢，單表查詢速度顯著提升，無olap數(shù)據(jù)庫的常見做法，通過寬表減少join帶來的性能消耗。

但是存在以下幾個問題：

• 雖然otter有不錯的封裝,通過數(shù)據(jù)路由能做一些簡單的數(shù)據(jù)拼接,但在調(diào)試上線復雜度上依然有不小的復雜度；
• otter偽裝mysql從庫同時要去做etl邏輯,把cdc干的活和實時ETL的活同時干了,耦合度較高
  。

2.2 實時架構1.0

2.2.1 flink+kafka+ClickHouse

在上述調(diào)研嘗試后都沒有解決根本的問題，我們開始把目標建立標準的實時數(shù)倉的思路上來，在20年olap沒有太多的可選項,我們把目標放在clickhouse上。

• 為了保證順序append每次寫入都會生成一個part文件，滿足一定條件后臺定時合并。
• 非常弱的update delete，不能保證原子性和實時性。
• clickhouse只適合數(shù)據(jù)量大，業(yè)務模型簡單,更新場景少的場景。
• 存算不分離，復雜查詢影響clickhouse寫入。

因為clickhouse的這些特性,尤其是不支持upsert的情況下,我們通常需要提前把大寬表的數(shù)據(jù)提前在flink聚合好,并且供應鏈數(shù)據(jù)生命周期長,作業(yè)流程也長如：

• 貨物的生命周期較短時長為一周，長周期時長超過1個月；
• 庫內(nèi)環(huán)節(jié)異常的多，從賣家發(fā)貨到收貨、分揀、質(zhì)檢、拍照、鑒別、防偽、復查、打包、出庫、買家簽收等十幾個甚至更多的環(huán)節(jié)，一張以商品實物id為主鍵的大寬表，需要join幾十張業(yè)務表；
• 供應鏈系統(tǒng)早期設計沒有每張表都會冗余唯一單號（入庫單，作業(yè)單，履約單）這樣的關鍵字段，導致沒辦法直接簡單的join數(shù)據(jù)。

在這樣一個架構下，們的flink在成本上，在穩(wěn)定性維護上，調(diào)優(yōu)上做的非常吃力。

附:

clickhouse不支持標準的upsert模式，可以通過使用AggregatingMergeTree 引擎字段類型使用SimpleAggregateFunction(anyLast, Nullable(UInt64)) 合并規(guī)則取最后一條非null數(shù)據(jù)可以實現(xiàn)upsert相似的功能，但讀時合并性能有影響。

2.3 實時架構2.0

2.3.1 flink+kafka+hologres

因此我們迫切的希望有支持upsert能力的olap數(shù)據(jù)庫,同時能搞定供應鏈寫多少的場景,也能搞定我們復雜查詢的場景,我們希望的olap數(shù)據(jù)至少能做到如下幾點：

• 有upsert能力，能對flink大任務做有效拆分；
• 存算分離，復雜業(yè)務計算，不影響業(yè)務寫入，同時能平滑擴縮容；
• 有一定的join能力帶來一些靈活度；
• 有完善的分區(qū)機制，熱數(shù)據(jù)查詢性能不受整體數(shù)據(jù)增長影響；
• 完善的數(shù)據(jù)備份機制。

這樣一個行列混合的olap數(shù)據(jù)庫,支持upsert,支持存算分離,還是比較符合我們的預期。

目前這樣一套架構支持了供應鏈每天數(shù)千人的報表取數(shù)需求,以及每天10億數(shù)據(jù)量的導出,訪問量在得物所有to B系統(tǒng)中排名靠前。

2.3.2 我們遇到的一些問題

多時間問題

如何設置segment_key,選擇哪個業(yè)務字段作為segment_key供應鏈幾十個環(huán)節(jié)都有操作時間,在不帶segment_key的情況下性能如何保障，困擾了我們一段時間。

設置合理的segment_key如有序的時間字段,可以做到完全順序?qū)?。每個segment文件都有個min,max值,所有的時間字段過來只需要去比較下在不在這個最小值最大值之間(這個動作開銷很低),不在范圍內(nèi)直接跳過，在不帶segment_key查詢的條件下,也能極大的降低所需要過濾的文件數(shù)量。

批流融合

背景：業(yè)務快速發(fā)展過程中，持續(xù)迭代實時任務成為常態(tài)。供應鏈業(yè)務復雜，環(huán)節(jié)多，流程往往長達一個月周期之久，這就導致state ttl設置周期長。job的operator變化（sql修改），checkpoint無法自動恢復，savepoint恢復機制無法滿足，比如增加group by和join。重新消費歷史數(shù)據(jù)依賴上游kafka存儲時效，kafka在公司平臺一般默認都是存儲7天，不能滿足一個月數(shù)據(jù)回刷需求場景。

方案：通過批流融合在source端實現(xiàn)離線 + 實時數(shù)據(jù)進行數(shù)據(jù)讀取、補齊。

（1）離線按key去重，每個key只保留一條，減少消息量下發(fā)。

（2）離線和實時數(shù)據(jù)合并，使用last_value取相同主鍵最新事件時間戳的一條數(shù)據(jù)。

（3）使用union all + group by方式是可作為代替join的一個選擇。

（4）實時數(shù)據(jù)取當日數(shù)據(jù)，離線數(shù)據(jù)取歷史數(shù)據(jù)，防止數(shù)據(jù)漂移，實時數(shù)據(jù)需前置一小時。

Join算子亂序

• 問題分析

由于join算子是對join鍵做hash后走不同的分片處理數(shù)據(jù)，開啟了2個并發(fā)后，再因為header_id字段的值變化，detail表2次數(shù)據(jù)流走到了2個不同的taskmanage,而不同的線程是無法保證輸出有序性的，所以數(shù)據(jù)有一定的概率會亂序輸出，導致期望的結(jié)果不正確，現(xiàn)象是數(shù)據(jù)丟失。

• 解決辦法

通過header inner join detail表后，拿到detail_id，這樣再次通過detail_id join就不會出現(xiàn)（join鍵）的值會從null變成非null的情況發(fā)生了，也就不會亂序了。

insert into sink
Select detail.id,detail.header_id,header.id
from detail
left join (
    Select detail.id AS detail_id,detail.header_id,header.id
    from header 
    inner join detail
    on detail.header_id  =  header.id 
) headerNew
on detail.id  =  headerNew.detail_id

2.3.3 Hologres or starrocks

這里也聊聊大家比較關注的hologres和starrocks,starrocks從開源開始也和我們保持了密切聯(lián)系,也做了多次的深入交流,我們也大致列了兩者之間的一些各自優(yōu)勢和對于我們看來一些不足的地方。

3、其他做的一些事情

3.1 開發(fā)提效工具——flink代碼生成器

參考MyBatis gennerator一些思想，利用模板引擎技術，定制化模板來生成flink sql。可以解決代碼規(guī)范，和提升開發(fā)效率?；究梢酝ㄟ^代碼配置來生成flink sql。

3.2 開發(fā)提效工具——可視化平臺

直接通過配置的方式,在線寫sql,直接生成頁面和接口,一鍵發(fā)布,同時引入緩存,鎖排隊機制解決高峰訪問性能問題。

動態(tài)配置接口，一鍵生成rpc服務:

動態(tài)配置報表:

4、未來規(guī)劃

當前架構依然存在某種程度的不可能三角，我們需要探索更多的架構可能性：

（1）利用寫在holo,計算在mc避免holo這種內(nèi)存數(shù)據(jù)庫，在極端查詢內(nèi)存被打爆的問題，利用mc的計算能力可以搞定一些事實表join的問題提升一些靈活度。

（2） 借助apache hudi推進湖倉一體，hudi做批流存儲統(tǒng)一，flink做批流計算統(tǒng)一，一套代碼，提供5-10分鐘級的準實時架構，緩解部分場景只需要準時降低實時計算成本。