個性化離線實時分析系統pora -- 系統架構 -- IT技術博客大學習 -- 共學習 共進步！

??? 個性化離線實時分析系統pora
????? 歡迎您，來自新浪微博的朋友！如果喜歡，您也可以把這篇文章分享到新浪微博：，這樣除了作為收藏還分享給了您的朋友！您也可以通過RSS訂閱來獲取我們的文章：Goolge源（推薦）, FeedSky源。
????? 祝您好運！

??????? 1.業務場景

?????? 伴隨著市場和技術的發展，個性化已經成為淘寶搜索的一個重要目標。簡單來說，個性化就是讓每個用戶在使用淘寶搜索時都能夠獲取自己最想要的結果，而不再是千篇一律的展示。實現個性化最直接的手段就是通過分析用戶的歷史行為日志，為用戶打上不同的標簽，在搜索中根據這些標簽來展示最貼近的結果。

?????? 在淘寶，用戶屬性分析是通過每天在云梯上定時運行的map reduce job來完成的，產出結果導入我們的在線kv存儲ups中，搜索引擎通過查詢ups獲取用戶屬性來為用戶返回個性化的結果。在云梯上執行的全量計算能夠進行復雜的模型計算，并且由于利用了云梯強大的計算能力，計算全部用戶幾十天的日志也只需花費幾個小時。

?????? 全量計算的不足之處在每次計算的輸入數據都是前一天到前N天的日志，無法將用戶當天的行為考慮進去，因此得到的用戶屬性永遠是滯后一天的，無法將某些用戶當前的屬性很好地反映出來。實時增量彌補了這一空缺，通過實時分析用戶的行為日志，將最新的用戶屬性反饋給搜索引擎，能夠為用戶展現最貼近其當前需求的結果。

??????? 2.系統需求

?????? 結合我們的業務場景和現狀，對實時分析系統大致有以下幾點需求。

?????? (1)不影響在線查詢的效率。這是一個最基本的需求，也決定了我們系統的定位：離線分析，將相對較重的分析過程放在離線階段完成，在線過程只需要查詢離線計算產出的結果即可。

?????? (2)實時。既然稱為實時系統，這也是一個起碼的要求。至于要多實時，初步的目標是從用戶一次行為發生到最后的屬性更新在幾秒內完成。

?????? (3)可水平擴展。個性化是一個需要長期打磨的系統工程，在不同的階段對系統的容量自然有不同的需求，這就需要我們的系統能夠具備良好的水平擴展能力。

?????? (4)能應對復雜多變的業務。算法同學會在個性化方面做各種嘗試，我們系統需要提供便利的方式來支持這些嘗試，最好是能夠將相對公用的東西與具體的業務邏輯剝離開，簡單來說，就是算法邏輯插件化。

?????? (5)高效。實時分析每天需要處理的日志量是巨大的，但是在其業務價值沒有得到足夠證明之前，是不可能占用太多的機器資源的，因此高效也成為了我們的一個基本需求。

??? 3.系統架構

?????? 說到實時分析，前提是實時日志收集，這方面淘寶已經有了一套的強大的日志收集和分發系統-TimeTunnel，俗稱TT，TT的延遲在幾百毫秒以內，并且提供根據游標來取消息的功能，基本滿足了我們消息對消息實時性和完整性的需求。全量計算的輸出是實時分析系統的另一個重要的數據源，因為我們寫入到ups提供給搜索引擎的是用戶屬性的最終結果，合并全量和增量的過程需要在實時分析系統中完成。全量計算是在云梯上完成的，結果存放在hdfs中，hdfs不能夠提供記錄級別的操作，考慮到我們的系統需求，必須要有另外一個提供高效的記錄級操作的存儲系統來保存這些數據。此外，由于算法邏輯通常會將用戶近兩天的行為都考慮進去，我們還需要保存用戶近期的行為記錄。我們選擇hbase作為全量結果和近期行為數據的存儲介質，一是由于hbase具有良好的水平擴展性，二是由于我們對hbase的使用比較熟悉。在計算系統的選型上，我們選擇了人見人愛的開源系統storm.各個組件的選型確定，整個系統的架構也就出來了。

?????? 系統架構

?????? (1)全量數據的導入。首先通過distcp方式將云梯上的數據拷貝到我們的hadoop集群中，然后使用bulk-load方式將數據導入到hbase表中。bulk-load是hbase提供的一種高效的數據批量導入工具，具體使用方法可以參考 http://hbase.apache.org/book/arch.bulk.load.html。 全量導入過程每天運行一次，我們會根據日期新建對應的表。

?????? (2)全量數據的切換和刪除。為了讓運行在storm中的實時分析拓撲檢測并使用到新全量表，我們另外創建了一張全量數據索引表，每次導入到新的全量數據表時更新對應的索引，實時分析拓撲定期掃描索引，在檢測到索引更新時自動切換到使用新表。

?????? (3)消息完整性的保證。實時分析拓撲中會保存消息處理的游標，并定期刷入到hbase中，這樣即使在節點失敗或者拓撲重啟的情況下也能夠恢復游標，處理堆積的消息。

??? 4.實時分析拓撲

?????? 當一條日志進入pora系統后，首先通過解析器解析出若干字段，然后通過過濾邏輯來判斷該條日志是否需要進行分析，如果需要，則會根據這些字段執行需要的join操作，例如將用戶、寶貝的信息補全，然后將join好的日志以及用戶的近期行為和全量屬性傳遞給系統中的算法插件，依次進行分析，最后將最新的用戶屬性更新到ups中，提供給外部使用。分析流程對應于storm的拓撲結構大致如下：

?????? (1)parser. 負責解析日志，根據配置文件取出需要的字段來。

?????? (2)filter. 過濾邏輯，根據某些規則過濾掉一些不感興趣的用戶日志。

?????? (3)joiner. 日志中的字段往往不能夠提供完整的信息，需要一個join過程來補全字段。在當前的實現中，我們會根據日志中的”行為”字段來使用不同的join方式。

?????? (4)analyzer. 主體分析邏輯。我們將這部分做成了一個 framework + plugins 的結構，其中framework負責取全量屬性、取近期行為、取當前行為，合并計算結果。每個plugin只需要實現analyze(全量屬性 + 近期行為 + 當前行為)的方法。framework對用戶屬性進行了字段切分，每個plugin只需要關心自己處理的那個字段即可。

?????? 在joiner和analyzer階段，我們做了一個很小的批量處理，不一定每條日志都會觸發計算，只有當累積夠一定條數后，才做一次集中處理，這樣在latency方面會有一些損失，但是能夠將對hbase的訪問打包，提高hbase的讀寫性能，從而大大提高系統的qps.這個批量的大小是可配的，用戶可以根據場景選擇配置，在qps和latency之間做trade-off,當配置為1的時候，就是完全的單條計算。

?????? (5)updater.負責將analyzer計算后發生更新的用戶屬性發送到ups中，繼而提供給搜索引擎使用。

??????? 5.系統監控

?????? 監控是一個線上系統必不可少的一部分。我們除了使用了一些基礎的機器狀態監控外，hbase集群還使用了集團hbase團隊開發的專用監控系統，非常直觀。此外，我們還需要一些業務指標的監控，例如我們的qps,latency,gap(日志處理時間與日志生產時間質檢單 間隔)，這方面也花費了我們一些心思。例如latency的監控，storm ui本身提供了即時數字的顯示，但是沒有我們想要的曲線圖(或許0.9版本中會有吧)。最后我們選擇了基于hbase的監控繪圖工具openTSDB。我們通過借助storm的ack機制來統計消息處理的latency,打印到日志中，然后使用一個腳本來搜集這些信息發送給openTSDB服務器來展示曲線。

?????? pora目前在淘寶個性化搜索中穩定運行，每天處理幾十億的日志信息，平均延遲在秒級。

??????? 6.經驗教訓

?????? (1).zookeeper集群獨立。因為zookeeper無論對于hbase還是storm都是至關重要的，最好將其單獨搞一個負載較低的集群。

?????? (2).hbase表的預分區。盡量將請求分散到各個節點上，至于預分區的原則，就根據業務場景來制定了。例如我們在存儲用戶全量屬性數據時是按照用戶名做哈希取模的。

?????? (3).storm使用經驗

?????? (a).根據需要修改默認參數。這點是顯然的,storm的默認參數并不能符合每個業務場景的需要，在storm源碼中的conf/defaults.xml目錄下有各個參數的默認取值，用戶可根據需要修改。

?????? (b).emit tuple時一定要new list.出于效率的考慮，storm底層的發送線程不會對該list進行深拷貝，會直接使用。如果用戶不小心修改了該list,會導致一些莫名其妙的失誤。

?????? (c).重啟supervopior前刪除本地data目錄。storm的supervisor會在本地data目錄保存一些狀態信息，在某些情況下這些狀態與zk中的最新狀態并不能保持一致，如果不刪除data目錄，容易導致supervisor重啟失敗。

個性化離線實時分析系統pora