訂單實(shí)時(shí)統(tǒng)計(jì)是其中一個(gè)典型的業(yè)務(wù)，對(duì)數(shù)據(jù)準(zhǔn)確性、性能等方面都有較高要求，也是上線時(shí)間最久的一個(gè)實(shí)時(shí)計(jì)算應(yīng)用。通過(guò)訂單實(shí)時(shí)統(tǒng)計(jì)，描述使用storm時(shí)，遇到的準(zhǔn)確性、性能、可靠性等方面的問(wèn)題。
訂單實(shí)時(shí)統(tǒng)計(jì)的演進(jìn)
第一版：流程走通
在使用storm之前，顯示實(shí)時(shí)統(tǒng)計(jì)數(shù)據(jù)一般有兩種方案：
1.在數(shù)據(jù)庫(kù)里執(zhí)行count、sum等聚合查詢，是簡(jiǎn)單快速的實(shí)現(xiàn)方案，但容易出現(xiàn)慢查詢。
2.在業(yè)務(wù)代碼里對(duì)統(tǒng)計(jì)指標(biāo)做累加，可以滿足指標(biāo)的快速查詢，但統(tǒng)計(jì)邏輯耦合到業(yè)務(wù)代碼，維護(hù)不方便，而且錯(cuò)誤數(shù)據(jù)定位和修正不方便。
既要解耦業(yè)務(wù)和統(tǒng)計(jì)，也要滿足指標(biāo)快速查詢，基于storm的實(shí)時(shí)計(jì)算方案可以滿足這兩點(diǎn)需求。
一個(gè)storm應(yīng)用的基本結(jié)構(gòu)有三部分：數(shù)據(jù)源、storm應(yīng)用、結(jié)果集。storm應(yīng)用從數(shù)據(jù)源讀取數(shù)據(jù)，經(jīng)過(guò)計(jì)算后，把結(jié)果持久化或發(fā)送消息給其他應(yīng)用。
第一版的訂單實(shí)時(shí)統(tǒng)計(jì)結(jié)構(gòu)如下圖。在數(shù)據(jù)源方面，最早嘗試在業(yè)務(wù)代碼里打日志的方式，但總有業(yè)務(wù)分支無(wú)法覆蓋，采集的數(shù)據(jù)不全。我們的業(yè)務(wù)數(shù)據(jù)庫(kù)是mysql，隨后嘗試基于mysql binlog的數(shù)據(jù)源，采用了阿里開源的canal，可以做到完整的收集業(yè)務(wù)數(shù)據(jù)變更。
在結(jié)果數(shù)據(jù)的處理上，我們把統(tǒng)計(jì)結(jié)果持久化到了mysql，并通過(guò)另一個(gè)后臺(tái)應(yīng)用的RESTful API對(duì)外提供服務(wù)，一個(gè)mysql就可以滿足數(shù)據(jù)的讀寫需求。
為了提升實(shí)時(shí)統(tǒng)計(jì)應(yīng)用吞吐量，需要提升消息的并發(fā)度。spout里設(shè)置了消息緩沖區(qū)，只要消息緩沖區(qū)不滿，就會(huì)源源不斷從消息源canal拉取數(shù)據(jù)，并把分發(fā)到多個(gè)bolt處理。
第二版：性能提升
第一版的性能瓶頸在統(tǒng)計(jì)結(jié)果持久化上。為了確保數(shù)據(jù)的準(zhǔn)確性，把所有的統(tǒng)計(jì)指標(biāo)持久化放在一個(gè)數(shù)據(jù)庫(kù)事務(wù)里。一筆訂單狀態(tài)更新后，會(huì)在一個(gè)事務(wù)里有兩類操作：
1.訂單的歷史狀態(tài)也在數(shù)據(jù)庫(kù)里存著，要與歷史狀態(tài)對(duì)比決定統(tǒng)計(jì)邏輯，并把最新的狀態(tài)持久化。storm的應(yīng)用本身是無(wú)狀態(tài)的，需要使用存儲(chǔ)設(shè)備記錄狀態(tài)信息
2.當(dāng)大家知道實(shí)時(shí)計(jì)算好用后，各產(chǎn)品都希望有實(shí)時(shí)數(shù)據(jù)，統(tǒng)計(jì)邏輯越來(lái)越復(fù)雜。店鋪、商品、用戶等多個(gè)指標(biāo)的寫操作都是在一個(gè)事務(wù)里commit，這一簡(jiǎn)單粗暴的方式早期很好滿足的統(tǒng)計(jì)需求，但是對(duì)于update操作持有鎖時(shí)間過(guò)長(zhǎng)，嚴(yán)重影響了并發(fā)能力。
為此做了數(shù)據(jù)庫(kù)事務(wù)的瘦身：
1.去除歷史狀態(tài)的mysql持久化，而是通過(guò)單條binlog消息的前后狀態(tài)對(duì)比，決定統(tǒng)計(jì)邏輯，這樣就做到了統(tǒng)計(jì)邏輯上的無(wú)狀態(tài)。但又產(chǎn)生了新問(wèn)題，如何保證消息有且只有處理一次，為此引入了一個(gè)redis用于保存最近24小時(shí)內(nèi)已成功處理的消息binlog偏移量，而storm的消息分發(fā)機(jī)制又可以保證相同消息總是能分配到一個(gè)bolt，避免線程安全問(wèn)題。
2.統(tǒng)計(jì)業(yè)務(wù)拆分，先是線上業(yè)務(wù)和公司內(nèi)部業(yè)務(wù)分離，隨后又把線上業(yè)務(wù)按不同產(chǎn)品拆分。這個(gè)不僅僅是bolt級(jí)別的拆分，而是在spout就完全分開
3.隨著統(tǒng)計(jì)應(yīng)用拆分，在canal和storm應(yīng)用之間加上消息隊(duì)列。canal不支持多消費(fèi)者，而實(shí)時(shí)統(tǒng)計(jì)業(yè)務(wù)也不用關(guān)系數(shù)據(jù)庫(kù)底層遷移、主從切換等維護(hù)工作，加上消息隊(duì)列能把底層數(shù)據(jù)的維護(hù)和性能優(yōu)化交給更專業(yè)的團(tuán)隊(duì)來(lái)做。
4.熱點(diǎn)數(shù)據(jù)在mysql里做了分桶。比如，通常一個(gè)店鋪天級(jí)別的統(tǒng)計(jì)指標(biāo)在mysql里是一行數(shù)據(jù)。如果這個(gè)店鋪有突發(fā)的大量訂單，會(huì)出現(xiàn)多個(gè)bolt同時(shí)去update這行數(shù)據(jù)，出現(xiàn)數(shù)據(jù)熱點(diǎn)，mysql里該行數(shù)據(jù)的鎖競(jìng)爭(zhēng)異常激烈。我們把這樣的熱點(diǎn)數(shù)據(jù)做了分桶，實(shí)驗(yàn)證明在特定場(chǎng)景下可以有一個(gè)數(shù)量級(jí)吞吐量提升。
最終，第二版的訂單實(shí)時(shí)統(tǒng)計(jì)結(jié)構(gòu)如下，主要變化在于引入了MQ，并使用redis作為消息狀態(tài)的存儲(chǔ)。而且由最初的一個(gè)應(yīng)用，被拆成了多個(gè)應(yīng)用。
第三版：準(zhǔn)確性提升
經(jīng)過(guò)第二版的優(yōu)化，實(shí)時(shí)統(tǒng)計(jì)的吞吐量已經(jīng)不成問(wèn)題，但還是遇到了做大數(shù)據(jù)最重要的準(zhǔn)確性的問(wèn)題：
1.統(tǒng)計(jì)口徑是會(huì)變化的，同樣是GMV，一年前和現(xiàn)在的算法可能有變化。例如一筆貨到付款訂單，是買家下單算成交，還是賣家發(fā)貨成交，在不同的時(shí)期可能使用不同的算法。
2.實(shí)時(shí)統(tǒng)計(jì)只能按照當(dāng)時(shí)的算法來(lái)做計(jì)算。有可能出現(xiàn)一段時(shí)間周期內(nèi)的GMV，前一段是按舊算法來(lái)計(jì)算，后一段按新算法來(lái)計(jì)算，提供的數(shù)據(jù)就不準(zhǔn)確了。
3.實(shí)時(shí)統(tǒng)計(jì)難免會(huì)出現(xiàn)bug，有不準(zhǔn)確的結(jié)果，修復(fù)錯(cuò)誤數(shù)據(jù)是個(gè)難題。
為了解決這個(gè)問(wèn)題，凡是涉及到兩天以前數(shù)據(jù)的，一律由離線計(jì)算提供，最終展示給用戶的數(shù)據(jù)，就是歷史離線統(tǒng)計(jì)數(shù)據(jù)，并上今日昨日實(shí)時(shí)統(tǒng)計(jì)數(shù)據(jù)。為什么是今日昨日實(shí)時(shí)統(tǒng)計(jì)呢？因?yàn)殡x線統(tǒng)計(jì)有數(shù)據(jù)準(zhǔn)備、建模、統(tǒng)計(jì)的過(guò)程，要花費(fèi)幾個(gè)小時(shí)，每天的凌晨很可能還得不到前一天的離線統(tǒng)計(jì)結(jié)果。
一旦統(tǒng)計(jì)口徑有變化，只需要重跑離線統(tǒng)計(jì)任務(wù)就可修復(fù)歷史數(shù)據(jù)，做到了冷熱數(shù)據(jù)分離。
實(shí)時(shí)計(jì)算的常見問(wèn)題
通過(guò)訂單實(shí)時(shí)統(tǒng)計(jì)的案例，可以抽象出一些基于storm實(shí)時(shí)計(jì)算的共性問(wèn)題。
消息狀態(tài)管理
storm不提供消息狀態(tài)管理，而且為了達(dá)到水平擴(kuò)展，最好是消息之間無(wú)狀態(tài)。對(duì)于大數(shù)據(jù)量、低精度的應(yīng)用，需要做到無(wú)狀態(tài)。而像訂單實(shí)時(shí)統(tǒng)計(jì)這樣數(shù)據(jù)量不算太大，但精度要求極高的場(chǎng)景，需要記錄消息處理狀態(tài)。而為了應(yīng)付重啟、分布式擴(kuò)展的場(chǎng)景，往往需要額外的介質(zhì)來(lái)存儲(chǔ)狀態(tài)。狀態(tài)信息往往是kv形式的讀寫，我們?cè)趯?shí)際的應(yīng)用中，使用過(guò)redis、HBase作為存儲(chǔ)。
消息不丟失、不重復(fù)、不亂序
對(duì)于準(zhǔn)確性要求高的場(chǎng)景，需要保證數(shù)據(jù)正確的只消費(fèi)一次。storm的有三種消息處理模式：
1.at most once，若不實(shí)現(xiàn)ack和fail方法，無(wú)論后續(xù)處理結(jié)果如何，消息只會(huì)發(fā)送一次，必定不能滿足高準(zhǔn)確性；
2.at least once，若實(shí)現(xiàn)了ack和fail方法，只有調(diào)用了ack方法才會(huì)任務(wù)處理成功，否則會(huì)重試?？赡軙?huì)出現(xiàn)消息重復(fù)，在并發(fā)場(chǎng)景下重復(fù)又意味著可能出現(xiàn)亂序；
3.exactly once，trident每個(gè)micro batch作為整體只成功處理一次，但也是無(wú)法保證消息真的只正確的處理一次，比如數(shù)據(jù)已經(jīng)處理完畢并持久化，但向數(shù)據(jù)源ack時(shí)失敗，就可能會(huì)有重試。
對(duì)于消息重復(fù)、亂序的場(chǎng)景，不是簡(jiǎn)單的消息冪等能解決，有以下的處理思路：
1.使用前面提到的狀態(tài)管理的辦法，識(shí)別出重復(fù)、亂序的數(shù)據(jù)；
2.業(yè)務(wù)邏輯中，兼容重復(fù)、亂序數(shù)據(jù)，比如維護(hù)一個(gè)業(yè)務(wù)狀態(tài)機(jī)，把異常數(shù)據(jù)剔除。
對(duì)于時(shí)序判斷，盡量不用使用時(shí)間戳，因?yàn)樵诜植际较到y(tǒng)里，各服務(wù)器時(shí)間不一致是很常見的問(wèn)題。
我們會(huì)嘗試在運(yùn)行過(guò)程中重啟消息源、storm應(yīng)用、存儲(chǔ)/MQ等下游系統(tǒng)，或者制造網(wǎng)絡(luò)丟包、延遲等異常，手工觸發(fā)可能的消息丟失、重復(fù)、亂序場(chǎng)景，來(lái)驗(yàn)證我們的應(yīng)用能否對(duì)應(yīng)這些異常情況。
復(fù)雜拓?fù)?/strong>
在storm的文檔里，有很多類似下圖的復(fù)雜應(yīng)用。
對(duì)于需要消息可靠處理的場(chǎng)景，是不適合這樣復(fù)雜拓?fù)涞模糠质∪绾位貪L，是否要全部bolt處理完畢才ack，是需要面對(duì)的問(wèn)題。過(guò)長(zhǎng)的拓?fù)滏溌?，里面的慢速邏輯?huì)拖慢整體性能。
可以考慮使用更簡(jiǎn)化的拓?fù)?，不同的邏輯之間盡量解耦，需要使用bolt的結(jié)果時(shí)，可以把數(shù)據(jù)持久化或者推送到MQ。
監(jiān)控
生產(chǎn)環(huán)境少不了監(jiān)控，除了服務(wù)器的基礎(chǔ)監(jiān)控，還加了不少storm特有的監(jiān)控：
1.消息延遲：消息在業(yè)務(wù)系統(tǒng)的時(shí)間戳與storm應(yīng)用的當(dāng)前時(shí)間戳對(duì)比，大于一定閾值則告警，不同應(yīng)用的閾值會(huì)不同；
2.消息處理時(shí)長(zhǎng)、fail數(shù)：這兩個(gè)都可以由storm的接口獲取，數(shù)值偏大很可能是出了問(wèn)題；
3.應(yīng)用TPS：記錄應(yīng)用的emit、ack、fail數(shù)的變化趨勢(shì)，幫助分析應(yīng)用的運(yùn)行情況；
4.任務(wù)級(jí)監(jiān)控：每臺(tái)服務(wù)器的worker、executor數(shù)量，這也可以通過(guò)storm接口獲取。
除此之外，會(huì)有各類應(yīng)用特有的監(jiān)控，一般都是離線計(jì)算的結(jié)果與實(shí)時(shí)計(jì)算結(jié)果對(duì)比。對(duì)于數(shù)據(jù)同步類的應(yīng)用，數(shù)據(jù)量比較大，可能會(huì)使用采樣的方式做校驗(yàn)。
后記
最近spark streaming、Flink等其他實(shí)時(shí)計(jì)算框架也很火，出于技術(shù)棧的維護(hù)成本的考慮，我們并沒有過(guò)多使用新的技術(shù)，太多框架一起維護(hù)不是件容易的事。
基于storm的實(shí)時(shí)計(jì)算應(yīng)用開發(fā)有幾個(gè)痛點(diǎn)：
1.消息重復(fù)、亂序的解決起來(lái)很麻煩，在分布式系統(tǒng)里很難有通用的辦法，而類似Flink提供通用邏輯封裝，是一種問(wèn)題解決的思路；
2.不能統(tǒng)一實(shí)時(shí)和離線開發(fā)，spark的統(tǒng)一不徹底，最新的apache beam則在API層做了統(tǒng)一封裝，也不能根本解決這個(gè)問(wèn)題；
3.高吞吐量場(chǎng)景，micro-batch的模式，會(huì)比流式模式有優(yōu)勢(shì)，我們?cè)谝恍┤罩咎幚韴?chǎng)景下使用了spark streaming。
后續(xù)我們會(huì)考慮將實(shí)時(shí)計(jì)算平臺(tái)化，解決或減輕上述幾個(gè)痛點(diǎn)，降低開發(fā)和維護(hù)成本。
閱讀排行榜
