1、背景

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

1.1 業(yè)務(wù)早期

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

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

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

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

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

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

2、架構(gòu)演變

2.1 原始階段

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

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

2.1.2 通過Otter完成大寬表的建設(shè)

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

但是存在以下幾個(gè)問題：

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

2.2 實(shí)時(shí)架構(gòu)1.0

2.2.1 flink+kafka+ClickHouse

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

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

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

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

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

附:

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

2.3 實(shí)時(shí)架構(gòu)2.0

2.3.1 flink+kafka+hologres

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

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

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

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

2.3.2 我們遇到的一些問題

多時(shí)間問題

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

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

批流融合

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

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

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

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

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

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

Join算子亂序

• 問題分析

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

• 解決辦法

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

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

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

3、其他做的一些事情

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

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

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

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

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

動(dòng)態(tài)配置報(bào)表:

4、未來規(guī)劃

當(dāng)前架構(gòu)依然存在某種程度的不可能三角，我們需要探索更多的架構(gòu)可能性：

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

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