背景

目前美團(tuán)點(diǎn)評(píng)已累計(jì)了豐富的線上交易與用戶行為數(shù)據(jù)，為商家賦能需要我們有更強(qiáng)大的專業(yè)化數(shù)據(jù)加工能力，來(lái)幫助商家做出正確的決策從而提高用戶體驗(yàn)。目前商家端產(chǎn)品在數(shù)據(jù)應(yīng)用上主要基于離線數(shù)據(jù)加工，數(shù)據(jù)生產(chǎn)調(diào)度以“T+1”為主，伴隨著越來(lái)越深入的精細(xì)化運(yùn)營(yíng)，實(shí)時(shí)數(shù)據(jù)應(yīng)用訴求逾加強(qiáng)烈。本文將從目前主流實(shí)時(shí)數(shù)據(jù)處理引擎的特點(diǎn)和我們面臨的問(wèn)題出發(fā)，簡(jiǎn)單的介紹一下我們是如何搭建實(shí)時(shí)數(shù)據(jù)處理系統(tǒng)。

設(shè)計(jì)框架

目前比較流行的實(shí)時(shí)處理引擎有 Storm，Spark Streaming，Flink。每個(gè)引擎都有各自的特點(diǎn)和應(yīng)用場(chǎng)景。 下表是對(duì)這三個(gè)引擎的簡(jiǎn)單對(duì)比：

考慮到每個(gè)引擎的特點(diǎn)、商家端應(yīng)用的特點(diǎn)和系統(tǒng)的高可用性，我們最終選擇了 Storm 作為本系統(tǒng)的實(shí)時(shí)處理引擎。

面臨的問(wèn)題

數(shù)據(jù)量的不穩(wěn)定性，導(dǎo)致對(duì)機(jī)器需求的不確定性。用戶的行為數(shù)據(jù)會(huì)受到時(shí)間的影響，比如半夜時(shí)刻和用餐高峰時(shí)段每分鐘產(chǎn)生的數(shù)據(jù)量有兩個(gè)數(shù)量級(jí)的差異。

上游數(shù)據(jù)質(zhì)量的不確定性。

數(shù)據(jù)計(jì)算時(shí)，數(shù)據(jù)的落地點(diǎn)應(yīng)該放到哪里來(lái)保證計(jì)算的高效性。

如何保證數(shù)據(jù)在多線程處理時(shí)數(shù)據(jù)計(jì)算的正確性。

計(jì)算好的數(shù)據(jù)以什么樣的方式提供給應(yīng)用方。

具體的實(shí)施方案

實(shí)時(shí)攝入數(shù)據(jù)完整性保障

數(shù)據(jù)完整性保證層：如何保證數(shù)據(jù)攝入到計(jì)算引擎的完整性呢？正如表格中比較的那樣，Storm 框架的語(yǔ)義為 At Least Once，至少攝入一次。這個(gè)語(yǔ)義的存在正好保證了數(shù)據(jù)的完整性，所以只需要根據(jù)自己的需求編寫(xiě) Spout 即可。好消息是我們的技術(shù)團(tuán)隊(duì)已經(jīng)開(kāi)發(fā)好了一個(gè)滿足大多數(shù)需求的 Spout，可以直接拿來(lái)使用。特別需要注意的一點(diǎn)，在數(shù)據(jù)處理的過(guò)程中需要我們自己來(lái)剔除已經(jīng)處理過(guò)的數(shù)據(jù)，因?yàn)?Storm 的語(yǔ)義會(huì)可能導(dǎo)致同一條數(shù)據(jù)攝入兩次。灰度發(fā)布期間（一周）對(duì)數(shù)據(jù)完整性進(jìn)行驗(yàn)證，數(shù)據(jù)完整性為100%。

實(shí)時(shí)數(shù)據(jù)平滑處理

數(shù)據(jù)預(yù)測(cè)層：實(shí)時(shí)的數(shù)據(jù)預(yù)測(cè)可以幫助我們對(duì)到達(dá)的數(shù)據(jù)進(jìn)行有效的平滑，從而可以減少在某一時(shí)刻對(duì)集群的壓力。 在數(shù)據(jù)預(yù)測(cè)方面，我們采用了在數(shù)學(xué)上比較簡(jiǎn)單的多元線性回歸模型（如果此模型不滿足業(yè)務(wù)需求，可以選用一些更高級(jí)別的預(yù)測(cè)模型），預(yù)測(cè)下一分鐘可能到來(lái)的數(shù)據(jù)的量。在數(shù)據(jù)延遲可接受的范圍內(nèi)，對(duì)數(shù)據(jù)進(jìn)行平滑，并完成對(duì)數(shù)據(jù)的計(jì)算。通過(guò)對(duì)該方案的使用，減輕了對(duì)集群約33%的壓力。具體步驟如下：

• 步驟一：將多個(gè)業(yè)務(wù)的實(shí)時(shí)數(shù)據(jù)進(jìn)行抽象化，轉(zhuǎn)換為(Y_i,X_1𝑖,X_2𝑖,X_3i,… ,X_ni)，其中Y_i為在(X_1i…X_ni)屬性下的數(shù)據(jù)量，(X_1i…X_ni)為n個(gè)不同的屬性，比如時(shí)間、業(yè)務(wù)、用戶的性別等等。
• 步驟二：因?yàn)榭紤]到實(shí)時(shí)數(shù)據(jù)的特殊性，不同業(yè)務(wù)的數(shù)據(jù)量隨時(shí)間變量基本呈現(xiàn)為M走勢(shì)，所以為了將非線性走勢(shì)轉(zhuǎn)換為線性走勢(shì)，可以將時(shí)間段分為4部分，保證在每個(gè)時(shí)間段內(nèi)數(shù)據(jù)的走勢(shì)為線性走勢(shì)。同理，如果其他的屬性使得走勢(shì)變?yōu)榉蔷€性，也可以分段分析。

• 步驟三：將抽象好的數(shù)據(jù)代入到多元線性回歸模型中，其方程組形式為：

  即：

  通過(guò)對(duì)該模型的求解方式求得估計(jì)參數(shù)，最后得多元線性回歸方程。

• 步驟四：數(shù)據(jù)預(yù)測(cè)完之后通過(guò)控制對(duì)數(shù)據(jù)的處理速度，保證在規(guī)定的時(shí)間內(nèi)完成對(duì)規(guī)定數(shù)據(jù)的計(jì)算，減輕對(duì)集群的壓力。

實(shí)時(shí)數(shù)據(jù)計(jì)算策略

策略層：Key/Value 模式更適應(yīng)于實(shí)時(shí)數(shù)據(jù)模型，不管是在存儲(chǔ)還是計(jì)算方面。Cellar（我們內(nèi)部基于阿里開(kāi)源的Tair研發(fā)的公共KV存儲(chǔ)）作為一個(gè)分布式的 Key/Value 結(jié)構(gòu)數(shù)據(jù)的解決方案，可以做到幾乎無(wú)延遲的進(jìn)行 IO 操作，并且可以支持高達(dá)千萬(wàn)級(jí)別的 QPS，更重要的是 Cellar 支持很多原子操作，運(yùn)用在實(shí)時(shí)數(shù)據(jù)計(jì)算上是一個(gè)不錯(cuò)的選擇。所以作為數(shù)據(jù)的落腳點(diǎn)，本系統(tǒng)選擇了Cellar。

但是在數(shù)據(jù)計(jì)算的過(guò)程中會(huì)遇到一些問(wèn)題，比如說(shuō)統(tǒng)計(jì)截止到當(dāng)前時(shí)刻入住旅館的男女比例是多少？很容易就會(huì)想到，從 Cellar 中取出截止到當(dāng)前時(shí)刻入住的男生是多少，女生是多少，然后做一個(gè)比值就 OK 了。但是本系統(tǒng)是在多線程的環(huán)境運(yùn)行的，如果該時(shí)刻有兩對(duì)夫婦入住了，產(chǎn)生了兩筆訂單，恰好這兩筆訂單被兩個(gè)線程所處理，當(dāng)線程A將該男士計(jì)算到結(jié)果中，正要打算將該女士計(jì)算到結(jié)果中的時(shí)候，線程B已經(jīng)計(jì)算完結(jié)果了，那么線程B計(jì)算出的結(jié)果就是2/1，那就出錯(cuò)啦。

所以為了保證數(shù)據(jù)在多線程處理時(shí)數(shù)據(jù)計(jì)算的正確性，我們需要用到分布式鎖。實(shí)現(xiàn)分布式鎖的方式有很多，本文就不贅述了。這里給大家介紹一種更簡(jiǎn)單快捷的方法。Cellar 中有個(gè) setNx 函數(shù)，該函數(shù)是原子的，并且是（Set If Not Exists），所以用該函數(shù)鎖住關(guān)鍵的字段就可以。就上面的例子而言，我們可以鎖住該旅館的唯一 ID 字段,計(jì)算完之后 delete 該鎖，這樣就可以保證了計(jì)算的正確性。

另外一個(gè)重要的問(wèn)題是 Cellar 不支持事務(wù)，就會(huì)導(dǎo)致該計(jì)算系統(tǒng)在升級(jí)或者重啟時(shí)會(huì)造成少量數(shù)據(jù)的不準(zhǔn)確。為了解決該問(wèn)題，運(yùn)用到一種 getset 原子思想的方法。如下：

public void doSomeWork(String input) {cellar.mapPut("uniq_ID");cellar.add("uniq_ID_1","some data");cellar.add("uniq_ID_2","some data again")；....cellar.mapRemove("uniq_ID"); }

如果上述代碼執(zhí)行到[2..5]某一行時(shí)系統(tǒng)重啟了，導(dǎo)致后續(xù)的操作并沒(méi)有完成，如何將沒(méi)有完成的操作添加上去呢？如下：

public void remedySomething() {map = cellar.mapGetAll();version = cellar.mapGet("uniq_ID").getVersion();for (string str : map) {if (cellar.get(str + "_1").getVersion()!= version) {cellar.add(str + "_1", "some data")；cellar.mapRemove(str);}.......} }

正如代碼里那樣，會(huì)有一個(gè)容器記錄了哪些數(shù)據(jù)正在被操作，當(dāng)系統(tǒng)重啟的時(shí)候，從該容器取出上次未執(zhí)行完的數(shù)據(jù)，用 Version（版本號(hào)）來(lái)記錄哪些操作還沒(méi)有完成，將沒(méi)有完成的操作補(bǔ)上，這樣就可以保證了計(jì)算結(jié)果的準(zhǔn)確性。起初 Version（版本號(hào)）被設(shè)計(jì)出來(lái)解決的問(wèn)題是防止由于數(shù)據(jù)的并發(fā)更新導(dǎo)致的問(wèn)題。

比如，系統(tǒng)有一個(gè) value 為“a,b,c”，A和B同時(shí)get到這個(gè) value。A執(zhí)行操作，在后面添加一個(gè)d，value 為 “a,b,c,d”。B執(zhí)行操作添加一個(gè)e，value為”a,b,c,e”。如果不加控制，無(wú)論A和B誰(shuí)先更新成功，它的更新都會(huì)被后到的更新覆蓋。Tair 無(wú)法解決這個(gè)問(wèn)題，但是引入了version 機(jī)制避免這樣的問(wèn)題。還是拿剛才的例子，A和B取到數(shù)據(jù)，假設(shè)版本號(hào)為10，A先更新，更新成 功后，value 為”a,b,c,d”，與此同時(shí)，版本號(hào)會(huì)變?yōu)?1。當(dāng)B更新時(shí)，由于其基于的版本號(hào)是10，服務(wù)器會(huì)拒絕更新，從而避免A的更新被覆蓋。B可以選擇 get 新版本的 value，然后在其基礎(chǔ)上修改，也可以選擇強(qiáng)行更新。

將 Version 運(yùn)用到事務(wù)的解決上也算是一種新型的使用。為驗(yàn)證該功能的正確性，灰度發(fā)布期間每天不同時(shí)段對(duì)項(xiàng)目進(jìn)行殺死并重啟，并對(duì)數(shù)據(jù)正確性進(jìn)行校驗(yàn)，數(shù)據(jù)的正確性為100%。

實(shí)時(shí)數(shù)據(jù)存儲(chǔ)

為了契合更多的需求，將數(shù)據(jù)分為三部分存儲(chǔ)。

Kafka：存儲(chǔ)稍加工之后的明細(xì)數(shù)據(jù)，方便做更多的擴(kuò)展。 MySQL：存儲(chǔ)中間的計(jì)算結(jié)果數(shù)據(jù)，方便計(jì)算過(guò)程的可視化。 Cellar：存儲(chǔ)最終的結(jié)果數(shù)據(jù)，供應(yīng)用層直接查詢使用。

應(yīng)用案例

美團(tuán)開(kāi)店寶的實(shí)時(shí)經(jīng)營(yíng)數(shù)據(jù)卡片。

美團(tuán)開(kāi)店寶作為美團(tuán)商家的客戶端，支持著眾多餐飲商家的輔助經(jīng)營(yíng)，而經(jīng)營(yíng)數(shù)據(jù)的實(shí)時(shí)性對(duì)影響商家決策尤為重要。該功能上線之后受到了商家的熱烈歡迎。卡片展示如下圖：

美團(tuán)點(diǎn)評(píng)金融合作門(mén)店的實(shí)時(shí)熱度標(biāo)簽。

該功能用于與美團(tuán)點(diǎn)評(píng)金融合作商家增加支付標(biāo)簽，用以突出這些商家，增加營(yíng)銷點(diǎn)。另一方面為優(yōu)質(zhì)商家吸引更多流量，為平臺(tái)帶來(lái)更多收益。展示如下圖：

總結(jié)與展望

以上就是該系統(tǒng)的設(shè)計(jì)框架與思路，并且部分功能已應(yīng)用到系統(tǒng)中。為了商家更好的決策，用戶更好的體驗(yàn)，在業(yè)務(wù)不斷增長(zhǎng)的情況下，對(duì)實(shí)時(shí)數(shù)據(jù)的分析就需要做到更全面。所以實(shí)時(shí)數(shù)據(jù)分析還有很多東西可以去做。

老生常談的大數(shù)據(jù) 4V+1O 特征，即數(shù)據(jù)量大(Volume)、類型繁多(Variety)、價(jià)值密度低(Value)、速度快時(shí)效性高(Velocity)、數(shù)據(jù)在線(Online)，相比離線數(shù)據(jù)系統(tǒng)，對(duì)實(shí)時(shí)數(shù)據(jù)的計(jì)算和應(yīng)用挑戰(zhàn)尤其艱巨。在技術(shù)框架演進(jìn)層面，對(duì)流式數(shù)據(jù)進(jìn)行高度抽象，簡(jiǎn)化開(kāi)發(fā)流程；在應(yīng)用端，我們后續(xù)希望在數(shù)據(jù)大屏、用戶行為分析產(chǎn)品、營(yíng)銷效果跟蹤等 DW/BI 產(chǎn)品進(jìn)行持續(xù)應(yīng)用，通過(guò)加快數(shù)據(jù)流轉(zhuǎn)的速度，更好的發(fā)揮數(shù)據(jù)價(jià)值。

參考

• 多元線性回歸模型

關(guān)于我們

到餐數(shù)據(jù)團(tuán)隊(duì)，用業(yè)內(nèi)最先進(jìn)的理念建設(shè)數(shù)據(jù)相關(guān)的系統(tǒng)和應(yīng)用，期待更多數(shù)據(jù)系統(tǒng)開(kāi)發(fā)、數(shù)據(jù)倉(cāng)庫(kù)開(kāi)發(fā)、數(shù)據(jù)建模好手的加入。 發(fā)郵件給liuqiang24@meituan.com 、xuyang14@meituan.com 、xuyang14@meituan.com。

總結(jié)

以上是生活随笔為你收集整理的美团点评基于Storm的实时数据处理实践的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。