基于Storm構建分布式實時處理應用初探

作者：唐潔

最近利用閑暇時間，又重新研讀了一下Storm。認真對比了一下Hadoop，前者更擅長的是，實時流式數(shù)據(jù)處理，后者更擅長的是基于HDFS，通過MapReduce方式的離線數(shù)據(jù)分析計算。對于Hadoop，本身不擅長實時的數(shù)據(jù)分析處理。兩者的共同點都是分布式架構，而且都類似有主/從關系的概念。

本文我不會具體闡述Storm集群和Zookeeper集群如何部署的問題，這里想通過一個實際的案例切入，分析一下如何利用Storm完成實時分析處理數(shù)據(jù)。

Storm本身是Apache托管的開源的分布式實時計算系統(tǒng)，它的前身是Twitter Storm。在Storm問世以前，處理海量的實時數(shù)據(jù)信息，大部分是類似于使用消息隊列，加上工作進程/線程的方式。這使得構建這類的應用程序，變得異常的復雜。很多的業(yè)務邏輯中，你不得不考慮消息的發(fā)送和接收，線程之間的并發(fā)控制等等問題。而其中的業(yè)務邏輯可能只是占據(jù)整個應用的一小部分，而且很難做到業(yè)務邏輯的解耦。但是Storm的出現(xiàn)改變了這種局面，它首先抽象出數(shù)據(jù)流Stream的抽象概念，一個Stream指的是tuples組成的無邊界的序列。后面又繼續(xù)提出Spouts、Bolts的概念。Spouts在Storm里面是數(shù)據(jù)源，專門負責生成流。而Bolts則是以流作為輸入，并重新生成流作為輸出，并且Bolts還會繼續(xù)指定它輸入的流應該如何劃分。最后Storm是通過拓撲（Topology）這種抽象概念，組織起若干個Spouts、Bolts構成的分布式數(shù)據(jù)處理網(wǎng)絡。Storm設計的時候，就有意的把Spouts、Bolts組成的拓撲（Topology）網(wǎng)絡通過Thrift服務方式進行封裝，這個做法，使得Storm的Spouts、Bolts組件可以通過目前主流的任意語言實現(xiàn)，使得整個框架的兼容性和擴展性更加優(yōu)秀。

在Storm里面拓撲（Topology）的概念，非常類似Hadoop里面MapReduce的Job的概念。不同的是Storm的拓撲（Topology）只要你啟動了，它就會一直運行下去，除非你kill掉；而MapReduce的Job最終它是會結束的?；谶@樣的模式，使得Storm非常適合處理實時性的數(shù)據(jù)分析、持續(xù)計算、DRPC（分布式RPC）等。

下面就結合實際的案例，設計分析一下，如何利用Storm改善應用的處理性能。

某通信公司的垃圾短信監(jiān)控平臺，實時地上傳每個省的疑似垃圾短信用戶的垃圾短信內(nèi)容文件，每個省則根據(jù)文件中垃圾短信的內(nèi)容，解析過濾出，包含指定敏感關鍵字的垃圾短信進行入庫。被入庫的垃圾短信用戶被列為敏感用戶，是重點監(jiān)控對象，畢竟亂發(fā)這些垃圾短信是非常不對的。垃圾短信監(jiān)控平臺生成的文件速度非常驚人，原來的傳統(tǒng)做法是，根據(jù)每個省的每一個地市，對應一個獨立應用，串行化地解析、過濾敏感關鍵字，來進行入庫處理。但是，從現(xiàn)狀來看，程序處理的性能并不高效，常常造成文件積壓，沒有及時處理入庫。

現(xiàn)在，我們就通過Storm來重新梳理、組織一下上述的應用場景。

首先，我先說明一下，該案例中Storm集群和Zookeeper集群的部署情況，如下圖所示：

Nimbus對應的主機是192.168.95.134是Storm主節(jié)點，其余兩臺從節(jié)點Supervisor對應的主機分別是192.168.95.135（主機名：slave1）、192.168.95.136（主機名：slave2）。同樣的，Zookeeper集群也是部署在上述節(jié)點上。

Storm集群和Zookeeper集群會互相通信，因為Storm就是基于Zookeeper的。然后先啟動每個節(jié)點的Zookeeper服務，其次分別啟動Storm的Nimbus、Supervisor服務。具體可以到Storm安裝的bin目錄下面啟動服務，啟動命令分別為storm nimbus > /dev/null 2 > &1 &和storm supervisor > /dev/null 2 > &1 &。然后用jps觀察啟動的效果。沒有問題的話，在Nimbus服務對應的主機上啟動Storm UI監(jiān)控對應的服務，在Storm安裝目錄的bin目錄輸入命令：storm ui >/dev/null 2>&1 &。然后打開瀏覽器輸入：http://{Nimbus服務對應的主機ip}:8080，這里就是輸入：http://192.168.95.134:8080/。觀察Storm集群的部署情況，如下圖所示：

可以發(fā)現(xiàn)，我們的Storm的版本是0.9.5，它的從節(jié)點（Supervisor）有2個，分別是slave1、slave2。一共的woker的數(shù)量是8個（Total slots）。Storm集群我們已經(jīng)部署完畢，也啟動成功了?，F(xiàn)在就利用Storm的方式，重新改寫一下這種敏感信息實時監(jiān)控過濾的應用。先看下Storm方式的拓撲結構圖：

其中的SensitiveFileReader-591、SensitiveFileReader-592（用戶短信采集器，分地市）代表的是Storm中的Spouts組件，表示一個數(shù)據(jù)的源頭，這里是表示從服務器的指定目錄下，讀取疑似垃圾短信用戶的垃圾短信內(nèi)容文件。當然Spouts的組件你可以根據(jù)實際的需求，擴展出許多Spouts。

然后讀取出文件中每一行的內(nèi)容之后，就是分析文件的內(nèi)容組件了，這里是指：SensitiveFileAnalyzer（監(jiān)控短信內(nèi)容拆解分析），它負責分析出文件的格式內(nèi)容。

為了簡單演示起見，我這里定義文件的格式為如下內(nèi)容（隨便寫一個例子）：home_city=591&user_id=5911000&msisdn=10000&sms_content=abc-slave1。每個列之間用&進行連接。其中home_city=591表示疑似垃圾短信的用戶歸屬地市編碼，591表示福州、592表示廈門；user_id=5911000表示疑似垃圾短信的用戶標識；msisdn=10000表示疑似垃圾短信的用戶手機號碼；sms_content=abc-slave1代表的就是垃圾短信的內(nèi)容了。SensitiveFileAnalyzer代表的就是Storm中的Bolt組件，用來處理Spouts“流”出的數(shù)據(jù)。

最后，就是我們根據(jù)解析好的數(shù)據(jù)，匹配業(yè)務規(guī)定的敏感關鍵字，進行過濾入庫了。這里我們是把過濾好的數(shù)據(jù)存入MySQL數(shù)據(jù)庫中。負責這項任務的組件是：SensitiveBatchBolt（敏感信息采集處理），當然它也是Storm中的Bolt組件。好了，以上就是完整的Storm拓撲（Topology）結構了。

現(xiàn)在，我們對于整個敏感信息采集過濾監(jiān)控的拓撲結構，有了一個整體的了解之后，我們再來看下如何具體編碼實現(xiàn)！先來看下整個工程的代碼層次結構，它如下圖所示：

首先來看下，我們定義的敏感用戶的數(shù)據(jù)結構RubbishUsers，假設我們要過濾的敏感用戶的短信內(nèi)容中，要包含“racketeer”、“Bad”等敏感關鍵字。具體代碼如下：

現(xiàn)在，我們看下敏感信息數(shù)據(jù)源組件SensitiveFileReader的具體實現(xiàn)，它負責從服務器的指定目錄下面，讀取疑似垃圾短信用戶的垃圾短信內(nèi)容文件，然后把每一行的數(shù)據(jù)，發(fā)送給下一個處理的Bolt（SensitiveFileAnalyzer），每個文件全部發(fā)送結束之后，在當前目錄中，把原文件重命名成后綴bak的文件（當然，你可以重新建立一個備份目錄，專門用來存儲這種處理結束的文件），SensitiveFileReader的具體實現(xiàn)如下：

監(jiān)控短信內(nèi)容拆解分析器SensitiveFileAnalyzer，這個Bolt組件，接收到數(shù)據(jù)源SensitiveFileReader的數(shù)據(jù)之后，就按照上面定義的格式，對文件中每一行的內(nèi)容進行解析，然后把解析完畢的內(nèi)容，繼續(xù)發(fā)送給下一個Bolt組件：SensitiveBatchBolt（敏感信息采集處理）。現(xiàn)在，我們來看下SensitiveFileAnalyzer這個Bolt組件的實現(xiàn)：

最后一個Bolt組件SensitiveBatchBolt（敏感信息采集處理）根據(jù)上游Bolt組件SensitiveFileAnalyzer發(fā)送過來的數(shù)據(jù)，然后跟業(yè)務規(guī)定的敏感關鍵字進行匹配，如果匹配成功，說明這個用戶，就是我們要重點監(jiān)控的用戶，我們把它通過hibernate采集到MySQL數(shù)據(jù)庫，統(tǒng)一管理。最后要說明的是，SensitiveBatchBolt組件還實現(xiàn)了一個監(jiān)控的功能，就是定期打印出，我們已經(jīng)采集到的敏感信息用戶數(shù)據(jù)?，F(xiàn)在給出SensitiveBatchBolt的實現(xiàn)：

由于是通過hibernate入庫到MySQL，所以給出hibernate配置，首先是：hibernate.cfg.xml

對應的ORM映射配置文件rubbish-users.hbm.xml內(nèi)容如下：

最后，還是通過Spring把hibernate集成起來，數(shù)據(jù)庫連接池用的是：DBCP。對應的Spring配置文件jdbc-hibernate-bean.xml的內(nèi)容如下：

到此為止，我們已經(jīng)完成了敏感信息實時監(jiān)控的所有的Storm組件的開發(fā)?，F(xiàn)在，我們來完成Storm的拓撲（Topology），由于拓撲（Topology）又分為本地拓撲和分布式拓撲，因此封裝了一個工具類StormRunner（拓撲執(zhí)行器），對應的代碼如下：

好了，現(xiàn)在我們把上面所有的Spouts/Bolts拼接成“拓撲”（Topology）結構，我們這里用的是分布式拓撲，來進行部署運行。具體的SensitiveTopology（敏感用戶監(jiān)控Storm拓撲）代碼如下：

到此為止，所有的Storm組件已經(jīng)開發(fā)完畢！現(xiàn)在，我們把上述工程打成jar包，放到Storm集群中運行，具體可以到Nimbus對應的Storm安裝目錄下面的bin目錄，輸入：storm jar + {jar路徑}。

比如我這里是輸入：storm jar /home/tj/install/SensitiveTopology.jar newlandframework.storm.topology.SensitiveTopology，然后，把疑似垃圾短信用戶的垃圾短信內(nèi)容文件放到指定的服務器下面的目錄（/home/tj/data/591、/home/tj/data/592），最后打開剛才的Storm UI，觀察任務的啟動執(zhí)行情況，這里如下圖所示：

可以看到我們剛才提交的拓撲：SensitiveTopology已經(jīng)成功提交到Storm集群里面了。這個時候，你可以鼠標點擊SensitiveTopology，然后會打開如下的一個Spouts/Bolts的監(jiān)控界面，如下圖所示：

我們可以很清楚地看到：Spouts組件（用戶短信采集器）：SensitiveFileReader591、SensitiveFileReader592的線程數(shù)executors、任務提交emitted情況。以及Bolts組件：監(jiān)控短信內(nèi)容拆解分析器（SensitiveFileAnalyzer）、敏感信息采集處理（SensitiveBatchBolt）的運行情況，這樣監(jiān)控起來就非常方便。

此外，我們還可以到對應的Supervisor服務器對應的Storm安裝目錄下面的logs目錄，查看一下worker的工作日志，我們來看下敏感信息監(jiān)控過濾的處理情況，截圖如下：

通過SensitiveBatchBolt模塊的監(jiān)控線程，可以看到，我們目前已經(jīng)采集到了9個敏感信息用戶了，再來看下，這些包含敏感關鍵字的用戶有沒有入庫MySQL成功？

發(fā)現(xiàn)入庫的結果也是9個，和日志打印的數(shù)量上是一致的。而且垃圾短信內(nèi)容sms_content果然都包含了“racketeer”、“Bad”這些敏感關鍵字！完全符合我們的預期。而且，以后文件處理量上來了，我們可以通過調(diào)整設置Spouts/Bolts的并行度，和Worker的數(shù)量進行化解。當然，你還可以通過水平擴展集群的數(shù)量來解決這個問題。

Storm在Apache開源項目的網(wǎng)址是：http://storm.apache.org/，有興趣的朋友可以經(jīng)常關注一下。官網(wǎng)上面有很權威的技術規(guī)范說明，以及如何把Storm和消息隊列、HDFS、HBase有效的集成起來。目前在國內(nèi)，就我個人看法，對Storm分析應用，做得最好的應該算是阿里巴巴，它在原來Storm的基礎上加以改良，開源出JStorm，有興趣的朋友，可以多關注一下。

借助Storm，我們可以很輕松地開發(fā)分布式實時處理應用，而上述場景的設計，只是Storm應用的一個案例。相比傳統(tǒng)的單機服務器應用而言，集群化地并行協(xié)同計算處理，是云計算、大數(shù)據(jù)時代的一個趨勢，也是我今后努力學習的方向。故在此寫下自己的學習經(jīng)驗體會，有不對的地方，還請各位群友批評指正。

生成海報

免責聲明：本網(wǎng)站內(nèi)容主要來自原創(chuàng)、合作伙伴供稿和第三方自媒體作者投稿，凡在本網(wǎng)站出現(xiàn)的信息，均僅供參考。本網(wǎng)站將盡力確保所提供信息的準確性及可靠性，但不保證有關資料的準確性及可靠性，讀者在使用前請進一步核實，并對任何自主決定的行為負責。本網(wǎng)站對有關資料所引致的錯誤、不確或遺漏，概不負任何法律責任。任何單位或個人認為本網(wǎng)站中的網(wǎng)頁或鏈接內(nèi)容可能涉嫌侵犯其知識產(chǎn)權或存在不實內(nèi)容時，應及時向本網(wǎng)站提出書面權利通知或不實情況說明，并提供身份證明、權屬證明及詳細侵權或不實情況證明。本網(wǎng)站在收到上述法律文件后，將會依法盡快聯(lián)系相關文章源頭核實，溝通刪除相關內(nèi)容或斷開相關鏈接。