基于Storm的滑動窗口實現(xiàn)

柴昱含　李道全

摘要：針對詐騙短信近年來出現(xiàn)的主叫號碼多變、被叫號碼具有隨機(jī)性、短信內(nèi)容難以識別等新特性，在綜合分析數(shù)據(jù)時，需要實時的處理海量的數(shù)據(jù)，而現(xiàn)有數(shù)據(jù)并發(fā)量大、實時性要求高等特點，所以采用什么樣的技術(shù)去處理這樣的數(shù)據(jù)十分關(guān)鍵。針對以上問題，對實時處理技術(shù)進(jìn)行調(diào)研。由于Storm在海量數(shù)據(jù)實時處理方面具有很好的特性，對Storm進(jìn)行相關(guān)調(diào)研。

關(guān)鍵詞：Storm；詐騙短信；大數(shù)據(jù)；Topology；實時流

中圖分類號：TP311 文獻(xiàn)標(biāo)識碼：A 文章編號：1009-3044（2014）16-3768-06

Abstract： In recent years， in the area of SMS scam， the calling numbers are highly changeable， the called numbers are of great randomicity， and the contents are unidentifiable. With these new features， an enormously great amount of data should be processed simultaneously when analyzing relevant data. Due to the great amount and the simultaneous nature， it is critical that what technique should be adopted to process these data. For the above issues， simultaneously processing technique is studied. As Storm boasts of brilliant capabilities in simultaneous mass data processing， in this paper Storm is investigated.

Key words： Storm ；SMS scam；mass dada；Topology；real-time streaming

實時數(shù)據(jù)流應(yīng)用的一個普通模式是對輸入數(shù)據(jù)進(jìn)行滾動計數(shù)，也被稱為滑動窗口分析.對于滾動計數(shù)的一個典型應(yīng)用是在一個用戶社區(qū)內(nèi)分析熱門話題-例如在Twitter-當(dāng)一個話題已經(jīng)在一個給定的時間窗口內(nèi)排名前N位時，則其已經(jīng)是一個熱門話題.本篇文章將介紹如何基于分布式和可擴(kuò)展的實時流數(shù)據(jù)處理平臺Storm實現(xiàn)這個算法.相同的代碼也能使用在其它領(lǐng)域例如安全監(jiān)控。

1 Storm簡介

1.1 Storm基本概念

Storm是一個分布式的實時處理系統(tǒng)，由主節(jié)點和從節(jié)點構(gòu)成。其中，主節(jié)點只有一個，并運行名為“Nimbus”的守護(hù)進(jìn)程；從節(jié)點有多個，每個工作節(jié)點都運行一個名為“Supervisor”的守護(hù)進(jìn)程?！癗imbus”進(jìn)程用于分配代碼、布置任務(wù)及故障檢測，“Supervisor”進(jìn)程用于監(jiān)聽工作，開始并終止工作進(jìn)程。Nimbus和Supervisor都能快速失敗，而且是無狀態(tài)的，這樣一來它們就變得十分健壯。主節(jié)點和從節(jié)點通過ZooKeeper來進(jìn)行交互，主節(jié)點通過ZooKeeper來發(fā)布指令，從節(jié)點從ZooKeeper讀取指令并執(zhí)行。ZooKeeper用于管理集群中的不同組件，ZeroMQ是內(nèi)部消息系統(tǒng)。

Storm的基本元素包括Topology、Stream、Spout、Bolt等。

Topology[2]：一個計算任務(wù)被稱為一個Topology，由多個Spout和Bolt組成。Topology任務(wù)一旦提交將會一直運行，除非主動停止任務(wù)，如圖2。

Stream：即數(shù)據(jù)流，是Storm中對數(shù)據(jù)的抽象，它是時間上無界的tuple元組序列。在Topology中，Spout是Stream的發(fā)射器，從特定數(shù)據(jù)源獲取數(shù)據(jù)發(fā)射Stream；一個Bolt可以接收多個Stream作為輸入，然后對數(shù)據(jù)進(jìn)行加工處理， Bolt還可以發(fā)射新的Stream給下級Bolt進(jìn)行處理。

Spout：消息源，可從外部獲取數(shù)據(jù)并將獲取的數(shù)據(jù)作為消息源提交給Topology。Spout包括可靠消息源和不可靠消息源兩類，可靠消息源將會對沒有被成功處理的tuple進(jìn)行重發(fā)，不可靠消息源不會重發(fā)。

Bolt：消息處理單元，可以執(zhí)行過濾， 聚合， 查詢等操作。

Topology[3]定義代碼示例：

TopologyBuilder builder = new TopologyBuilder（）；

builder.setSpout（“spout”， new KafkaSpout（），2）；

builder.setBolt（“smsBolt”， new SMSParerBolt （）， 8）.shuffleGrouping（“spout”）；

builder.setBolt（“filterBolt”， new FilterCallingNumberBolt （）， 2）

. shuffleGrouping （“smsBolt”）. setNumTasks（4）；

通過TopologyBuilder的setSpout（）方法設(shè)置Spout，示例中”spout”為Spout的別名，KafkaSpout為Spout的一個實現(xiàn)類，后面的數(shù)字2為并發(fā)線程數(shù)。通過setBolt（）方法設(shè)置Bolt，實例中分別設(shè)置了別名為”smsBolt”、線程并發(fā)度為8的SMSParerBolt和別名為”filterBolt”、線程并發(fā)度為2的FilterCallingNumberBolt，setBolt后的shuffleGrouping為流分組策略，setNumTasks為該Bolt對應(yīng)Task數(shù)。關(guān)于并發(fā)度、流分組策略、Task數(shù)的概念將在之后小節(jié)介紹。endprint

1.2 Storm流分組策略

流分組策略（StreamGrouping）[4]，用于設(shè)置Bolt的Task間數(shù)據(jù)的分配策略，包括以下幾類：

Shuffle Grouping：隨機(jī)分組，隨機(jī)派發(fā)stream里面的tuple，保證每個bolt接收到的tuple數(shù)目大致相同。

Fields Grouping：按字段分組，比如按userid來分組， 具有同樣userid的tuple會被分到相同的Bolts里的一個task， 而不同的userid則會被分配到不同的bolts里的task。

All Grouping：廣播發(fā)送，對于每一個tuple，所有的bolts都會收到。

Global Grouping：全局分組，這個tuple被分配到storm中的一個bolt的其中一個task。再具體一點就是分配給id值最低的那個task。

Non Grouping：不分組，這個分組的意思是說stream不關(guān)心到底誰會收到它的tuple。目前這種分組和Shuffle grouping是一樣的效果， 有一點不同的是storm會把這個bolt放到這個bolt的訂閱者同一個線程里面去執(zhí)行。

Direct Grouping：直接分組，這是一種比較特別的分組方法，用這種分組意味著消息的發(fā)送者指定由消息接收者的哪個task處理這個消息。 只有被聲明為Direct Stream的消息流可以聲明這種分組方法。而且這種消息tuple必須使用emitDirect方法來發(fā)射。消息處理者可以通過TopologyContext來獲取處理它的消息的task的id （OutputCollector.emit方法也會返回task的id）。

Local or shuffle grouping：如果目標(biāo)bolt有一個或者多個task在同一個工作進(jìn)程中，tuple將會被隨機(jī)發(fā)生給這些tasks。否則，和普通的Shuffle Grouping行為一致。

2 基于Storm的滑動窗口實現(xiàn)

2.1 熱門話題和滑動窗口

首先，解釋一下什么是熱門話題，以便于我們有一個共識。

2.1.1 熱門話題

一個單詞、短語或話題相比其它的標(biāo)簽有更多地被標(biāo)記的概率則被稱為熱門話題。成為熱門話題一方面來自于用戶的一致關(guān)注，另一方面因為某些特殊事件而引起人們的關(guān)注。這些話題有助于用戶了解當(dāng)前世界正在發(fā)生什么。

換一句話，它說明了在一個用戶社區(qū)內(nèi)“熱點是什么”。通常，你會對一個給定時間范圍內(nèi)的熱門話題感興趣，例如，在過去五分鐘內(nèi)或一天內(nèi)最流行的話題。所以，對于“熱點是什么”更準(zhǔn)確的描述應(yīng)該是“今天的熱點是什么”或“這周的熱點是什么”。

在這篇文章中，我們假設(shè)我們有一個系統(tǒng)，這個系統(tǒng)使用TwitterAPI去獲取最新的數(shù)據(jù)。更進(jìn)一步假設(shè)，我們有一種機(jī)制可以從Twitter的消息中以單詞的形式標(biāo)識主題。例如，我們可以選擇用一個簡單的模式匹配算法處理主題標(biāo)簽#。

我們設(shè)計我們的系統(tǒng)，如果一個主題A比主題B更多次數(shù)被提到，我們則認(rèn)為主題A比主題B更流行。這意味著我們只需要去統(tǒng)計Twitter中主題被引用的次數(shù)。

對于本文我們不關(guān)心這些主題是怎樣從用戶內(nèi)容或用戶活動中衍生出來的，只要知道這些衍生出來的主題用文本的方式表示就行了。然后，Storm的拓?fù)溥壿嫊ㄟ^滾動計算和排名計算分析出輸入數(shù)據(jù)的實時熱門話題。前期關(guān)注去對一定時間范圍內(nèi)用戶輸入數(shù)據(jù)的過濾，后期關(guān)注與對熱門話題的排名。

我們期望Storm拓?fù)溥壿嫸ㄆ诘漠a(chǎn)生TopN熱門話題，就像下面的輸出，t0-t2是不同的時間點：

Rank @ t0 ——> t1 ——> t2 —————————————————————— 1. java （33） ruby （41） scala （32） 2. php （30） scala （28） python （29） 3. scala （21） java （27） ruby （24） 4. ruby （16） python （21） java （21） 5. python （15） php （14） erlang （18）

在這個例子中我們可以看出“scala”已經(jīng)成為最熱門話題。

2.1.2 滑動窗口

在之前的背景介紹中，我想要說明的是滑動窗口即滾動計算。一張圖片勝過千言萬語：

在上述例子中，對滑動窗口內(nèi)的數(shù)據(jù)求和。

一個公式可能是很好的解釋：

從大小到時間：如果我們假設(shè)窗口每5分鐘前進(jìn)一次，則輸入數(shù)據(jù)中每個塊內(nèi)的數(shù)字代表過去相同時間間隔內(nèi)收集的數(shù)據(jù)量。在本例中，窗口大小是N*m分鐘。簡單的說，如果N=1和m=5，則我們的滑動窗口算法沒一分鐘提交過去五分鐘的數(shù)據(jù)。

現(xiàn)在我們已經(jīng)介紹過了熱門話題和時間窗口，我們最后來談?wù)勅绾瓮ㄟ^代碼實現(xiàn)。

2.2 滑動窗口的實現(xiàn)

2.2.1 實現(xiàn)數(shù)據(jù)結(jié)構(gòu)

接下來我們介紹一下核心數(shù)據(jù)結(jié)構(gòu)。正如你所看到的，一個有趣的特性是這些數(shù)據(jù)結(jié)構(gòu)與Storm的內(nèi)部特性是完全獨立的。我們的Storm bolts將要使用它們，當(dāng)然，數(shù)據(jù)結(jié)構(gòu)對于Storm沒有依賴。

計數(shù)所使用的類：SlotBasedCounter， SlidingWindowCounter

排名所使用的類：Rankings， Rankable， RankableObjectWithFields

另一個顯著的改善是代碼刪除了不必的代碼并且使用線程相關(guān)代碼，例如同步或手動啟動后臺線程。并且數(shù)據(jù)結(jié)構(gòu)也不與系統(tǒng)時間交互。消除直接調(diào)用系統(tǒng)時間并且手動開啟后臺線程是得新代碼比老代碼更簡單和容易測試。endprint

// such code from the old RollingCountObjects bolt is not needed anymore

long delta = millisPerBucket（_numBuckets） - （System.currentTimeMillis（） % millisPerBucket（_numBuckets））；

Utils.sleep（delta）；

SlotBasedCounter

SlotBasedCounter類提供了對于對象計數(shù)的功能。用于計數(shù)的Slots的數(shù)量是固定的。當(dāng)前類提供了四個公共方法：

public void incrementCount（T obj， int slot）；

public void wipeSlot（int slot）：

public long getCount（T obj， int slot）

public Map getCounts（）；

例子：

SlotBasedCounter counter = new SlotBasedCounter

Rankable rankable = RankableObjectWithFields.from（tuple）； super.getRankings（）.updateWith（rankable）；

}

仔細(xì)看一下Rankings， Rankable 和RankableObjectWithFields 。如果你自己不得不去實現(xiàn)這些類并且你是一個有經(jīng)驗的工程師，那么你一定會實現(xiàn)equals（） 和 hashCode（）方法。

2.2.4 實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

在閱讀這一部分的時候，“words”代表了我們所設(shè)想的系統(tǒng)中用戶提到的主題。

Rolling Top Words拓?fù)溥壿嬘蒚estWordSpout， RollingCountBolt， IntermediateRankingsBolt 和TotalRankingsBolt組成

滑動窗口的大小和提交頻率只是個例子，在我們例子中有一個5分鐘的滑動窗口并且每分鐘提交一次。

主要工作如下：

1）拓?fù)溥壿嫷牡谝粚覶estWordSpout模擬輸入數(shù)據(jù)-用戶提到的主題。

2）第二層RollingCountBolt對輸入數(shù)據(jù)進(jìn)行計數(shù)

3）第三層IntermediateRankingsBolt對數(shù)據(jù)進(jìn)行排名

4）最后，TotalRankingsBolt匯聚數(shù)據(jù)，輸出總排名。

拓?fù)溥壿嫶a如下：

builder.setSpout（spoutId， new TestWordSpout（）， 2）；

builder.setBolt（counterId， new RollingCountBolt（9， 3）， 3） .fieldsGrouping（spoutId， new Fields（"word"））；

builder.setBolt（intermediateRankerId， new IntermediateRankingsBolt（TOP_N）， 2） .fieldsGrouping（counterId， new Fields（"obj"））；

builder.setBolt（totalRankerId，new TotalRankingsBolt（TOP_N）） .globalGrouping（intermediateRankerId）；

參考文獻(xiàn)：

[1] 張春麟.手機(jī)垃圾短信過濾平臺的分析與應(yīng)用[D].北京：北京郵電大學(xué)，2010.

[2] 互聯(lián)網(wǎng)文檔資源.storm-0.8.2源碼分析之topology啟動[DB/OL].http：//blog.csdn.net/chlaws/article/details/10562035，2013.

[3] 互聯(lián)網(wǎng)文檔資源.Storm-源碼分析-Component，Executor，Task之間關(guān)系[DB/OL].http：//www.cnblogs.com/fxjwind/，2013.

[4] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].https：//github.com/nathanmarz/storm/wiki/Tutorial，2012.

[5] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].http：//storm.incubator.apache.org/documentation/Tutorial.html，2012.

[6] 互聯(lián)網(wǎng)文檔資源.使用Storm實現(xiàn)實時大數(shù)據(jù)分析[DB/OL].http：//www.csdn.net/article/2012-12-24/2813117-storm-realtime-big-data-analysis，2012.

[7] 互聯(lián)網(wǎng)文檔資源.Storm快速理解[DB/OL].http：//blog.csdn.net/colorant/article/details/8256039，2012.

[8] 互聯(lián)網(wǎng)文檔資源.徐明明.Twitter Storm： Transactional Topolgoy簡介[DB/OL].http：//xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/，2012.

[9] 互聯(lián)網(wǎng)文檔資源.Storm之trident聚合操作介紹[DB/OL].http：//blog.sina.com.cn/s/blog_6ff05a2c0101k6xj.html，2013.

[10] 互聯(lián)網(wǎng)文檔資源.[翻譯][Trident] Storm Trident教程[DB/OL].http：//blog.csdn.net/derekjiang/article/details/9126185，2013.endprint

Rankable rankable = RankableObjectWithFields.from（tuple）； super.getRankings（）.updateWith（rankable）；

}

仔細(xì)看一下Rankings， Rankable 和RankableObjectWithFields 。如果你自己不得不去實現(xiàn)這些類并且你是一個有經(jīng)驗的工程師，那么你一定會實現(xiàn)equals（） 和 hashCode（）方法。

2.2.4 實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

在閱讀這一部分的時候，“words”代表了我們所設(shè)想的系統(tǒng)中用戶提到的主題。

Rolling Top Words拓?fù)溥壿嬘蒚estWordSpout， RollingCountBolt， IntermediateRankingsBolt 和TotalRankingsBolt組成

滑動窗口的大小和提交頻率只是個例子，在我們例子中有一個5分鐘的滑動窗口并且每分鐘提交一次。

主要工作如下：

1）拓?fù)溥壿嫷牡谝粚覶estWordSpout模擬輸入數(shù)據(jù)-用戶提到的主題。

2）第二層RollingCountBolt對輸入數(shù)據(jù)進(jìn)行計數(shù)

3）第三層IntermediateRankingsBolt對數(shù)據(jù)進(jìn)行排名

4）最后，TotalRankingsBolt匯聚數(shù)據(jù)，輸出總排名。

拓?fù)溥壿嫶a如下：

builder.setSpout（spoutId， new TestWordSpout（）， 2）；

builder.setBolt（counterId， new RollingCountBolt（9， 3）， 3） .fieldsGrouping（spoutId， new Fields（"word"））；

builder.setBolt（intermediateRankerId， new IntermediateRankingsBolt（TOP_N）， 2） .fieldsGrouping（counterId， new Fields（"obj"））；

builder.setBolt（totalRankerId，new TotalRankingsBolt（TOP_N）） .globalGrouping（intermediateRankerId）；

參考文獻(xiàn)：

[1] 張春麟.手機(jī)垃圾短信過濾平臺的分析與應(yīng)用[D].北京：北京郵電大學(xué)，2010.

[2] 互聯(lián)網(wǎng)文檔資源.storm-0.8.2源碼分析之topology啟動[DB/OL].http：//blog.csdn.net/chlaws/article/details/10562035，2013.

[3] 互聯(lián)網(wǎng)文檔資源.Storm-源碼分析-Component，Executor，Task之間關(guān)系[DB/OL].http：//www.cnblogs.com/fxjwind/，2013.

[4] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].https：//github.com/nathanmarz/storm/wiki/Tutorial，2012.

[5] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].http：//storm.incubator.apache.org/documentation/Tutorial.html，2012.

[6] 互聯(lián)網(wǎng)文檔資源.使用Storm實現(xiàn)實時大數(shù)據(jù)分析[DB/OL].http：//www.csdn.net/article/2012-12-24/2813117-storm-realtime-big-data-analysis，2012.

[7] 互聯(lián)網(wǎng)文檔資源.Storm快速理解[DB/OL].http：//blog.csdn.net/colorant/article/details/8256039，2012.

[8] 互聯(lián)網(wǎng)文檔資源.徐明明.Twitter Storm： Transactional Topolgoy簡介[DB/OL].http：//xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/，2012.

[9] 互聯(lián)網(wǎng)文檔資源.Storm之trident聚合操作介紹[DB/OL].http：//blog.sina.com.cn/s/blog_6ff05a2c0101k6xj.html，2013.

[10] 互聯(lián)網(wǎng)文檔資源.[翻譯][Trident] Storm Trident教程[DB/OL].http：//blog.csdn.net/derekjiang/article/details/9126185，2013.endprint

Rankable rankable = RankableObjectWithFields.from（tuple）； super.getRankings（）.updateWith（rankable）；

}

仔細(xì)看一下Rankings， Rankable 和RankableObjectWithFields 。如果你自己不得不去實現(xiàn)這些類并且你是一個有經(jīng)驗的工程師，那么你一定會實現(xiàn)equals（） 和 hashCode（）方法。

2.2.4 實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

實現(xiàn)Rolling Top Words 拓?fù)溥壿?/p>

在閱讀這一部分的時候，“words”代表了我們所設(shè)想的系統(tǒng)中用戶提到的主題。

Rolling Top Words拓?fù)溥壿嬘蒚estWordSpout， RollingCountBolt， IntermediateRankingsBolt 和TotalRankingsBolt組成

滑動窗口的大小和提交頻率只是個例子，在我們例子中有一個5分鐘的滑動窗口并且每分鐘提交一次。

主要工作如下：

1）拓?fù)溥壿嫷牡谝粚覶estWordSpout模擬輸入數(shù)據(jù)-用戶提到的主題。

2）第二層RollingCountBolt對輸入數(shù)據(jù)進(jìn)行計數(shù)

3）第三層IntermediateRankingsBolt對數(shù)據(jù)進(jìn)行排名

4）最后，TotalRankingsBolt匯聚數(shù)據(jù)，輸出總排名。

拓?fù)溥壿嫶a如下：

builder.setSpout（spoutId， new TestWordSpout（）， 2）；

builder.setBolt（counterId， new RollingCountBolt（9， 3）， 3） .fieldsGrouping（spoutId， new Fields（"word"））；

builder.setBolt（intermediateRankerId， new IntermediateRankingsBolt（TOP_N）， 2） .fieldsGrouping（counterId， new Fields（"obj"））；

builder.setBolt（totalRankerId，new TotalRankingsBolt（TOP_N）） .globalGrouping（intermediateRankerId）；

參考文獻(xiàn)：

[1] 張春麟.手機(jī)垃圾短信過濾平臺的分析與應(yīng)用[D].北京：北京郵電大學(xué)，2010.

[2] 互聯(lián)網(wǎng)文檔資源.storm-0.8.2源碼分析之topology啟動[DB/OL].http：//blog.csdn.net/chlaws/article/details/10562035，2013.

[3] 互聯(lián)網(wǎng)文檔資源.Storm-源碼分析-Component，Executor，Task之間關(guān)系[DB/OL].http：//www.cnblogs.com/fxjwind/，2013.

[4] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].https：//github.com/nathanmarz/storm/wiki/Tutorial，2012.

[5] 互聯(lián)網(wǎng)文檔資源.Tutorial[DB/OL].http：//storm.incubator.apache.org/documentation/Tutorial.html，2012.

[6] 互聯(lián)網(wǎng)文檔資源.使用Storm實現(xiàn)實時大數(shù)據(jù)分析[DB/OL].http：//www.csdn.net/article/2012-12-24/2813117-storm-realtime-big-data-analysis，2012.

[7] 互聯(lián)網(wǎng)文檔資源.Storm快速理解[DB/OL].http：//blog.csdn.net/colorant/article/details/8256039，2012.

[8] 互聯(lián)網(wǎng)文檔資源.徐明明.Twitter Storm： Transactional Topolgoy簡介[DB/OL].http：//xumingming.sinaapp.com/736/twitter-storm-transactional-topolgoy/，2012.

[9] 互聯(lián)網(wǎng)文檔資源.Storm之trident聚合操作介紹[DB/OL].http：//blog.sina.com.cn/s/blog_6ff05a2c0101k6xj.html，2013.

[10] 互聯(lián)網(wǎng)文檔資源.[翻譯][Trident] Storm Trident教程[DB/OL].http：//blog.csdn.net/derekjiang/article/details/9126185，2013.endprint