Apache Storm技術參考手冊

目錄

簡介.............................................................................4

什么是ApacheStorm?.................................................................................................................4

ApacheStormvsHadoop...............................................................................................................4

使用ApacheStorm的例子.....................................................5

ApacheStorm優(yōu)勢............................................................5

Storm的核心組件.............................................................5

Topology...........................................................................................................................................8

Storm集群安裝..............................................................9

StormHelloWorld...........................................................................................................................9

環(huán)境準備................................................................9

具體流程................................................................10

核心概念........................................................................35

拓撲........................................................................36

任務........................................................................37

進程........................................................................37

流分組......................................................................37

隨機分組...............................................................37

字段分組...............................................................38

全局分組...............................................................39

所有分組...............................................................40

集群架構........................................................................40

工作流程........................................................................42

分布式消息系統(tǒng).................................................................44

什么是分布式消息系統(tǒng)？.....................................................44

Thrift協(xié)議...................................................................45

安裝............................................................................46

步驟1-驗證Java安裝.......................................................46

步驟1.1-下載JDK.............................................................................................................46

步驟1.2-解壓文件......................................................47

步驟1.3-移動到opt文件夾.............................................47

步驟1.4-設置路徑......................................................47

步驟1.5-Java替代項....................................................48

步驟1.6-Java安裝驗證..................................................48

第2步-ZooKeeper框架安裝.................................................48

步驟2.1-下載ZooKeeper.................................................................................................48

步驟2.2-解壓tar文件...................................................48

步驟2.3-創(chuàng)建配置文件..................................................49

步驟2.4-啟動ZooKeeper服務器..........................................49

步驟2.5?啟動CLI..............................................................................................................49

步驟2.6-停止ZooKeeper服務器..........................................50

第3步-ApacheStorm框架安裝...............................................51

步驟3.1-下載Storm.........................................................................................................51

步驟3.2??解壓tar文件...................................................51

步驟3.3-打開配置文件..................................................51

步驟3.4-啟動Nimbus.......................................................................................................52

步驟3.5-啟動Supervisor..................................................................................................52

步驟3.6-啟動UI...............................................................................................................52

工作實例........................................................................53

場景-移動呼叫日志分析器..................................................53

Spout創(chuàng)建..................................................................53

open.......................................................................................................................................54

nextTuple...............................................................................................................................54

close.......................................................................................................................................55

declareOutputFields.............................................................................................................55

ack..........................................................................................................................................55

fail...........................................................................................................................................55

FakeCallLogReaderSpout......................................................................................................56

編碼-FakeCallLogReaderSpout.java.................................................................................56

Bolt創(chuàng)建....................................................................60

Prepare...................................................................................................................................60

execute...................................................................................................................................60

cleanup...................................................................................................................................61

declareOutputFields.............................................................................................................61

呼叫日志創(chuàng)建者bolt...................................................................................................................61

編碼-CallLogCreatorBoltjava............................................................................................62

呼叫日志計數器Bolt...................................................................................................................64

編碼-CallLogCounterBolt.java..........................................................................................64

創(chuàng)建拓撲....................................................................66

本地集群....................................................................67

編碼-LogAnalyserStorm.java.............................................................................................67

構建和運行應用程序.........................................................69

輸出....................................................................69

非JVM語言.................................................................70

Python綁定.............................................................70

Trident....................................................................................................................................................71

Trident拓撲.................................................................72

TridentTuples................................................................................................................................72

TridentSpout.................................................................................................................................73

Trident操作.................................................................73

過濾....................................................................74

函數....................................................................乃

聚合....................................................................76

分組....................................................................78

合并和連接.............................................................79

狀態(tài)維護....................................................................79

分布式RPC...................................................................................................................................80

什么時候使用Trident?.............................................................................................................80

Trident的工作實例...........................................................80

格式化呼叫信息.........................................................81

編碼：FormatCall.java.........................................................................................................81

CSVSplit.................................................................................................................................81

編碼：CSVSplit.java.............................................................................................................82

日志分析器.............................................................82

編碼：LogAnalyserTrident.java..........................................................................................83

構建和運行應用程序.........................................................86

輸出....................................................................87

在Twitter上的應用..............................................................88

Twitter............................................................................................................................................88

Spout創(chuàng)建..............................................................88

編碼：TwitterSampleSpout.java........................................................................................89

Hashtag閱讀器spout..................................................................................................................94

編碼：HashtagReaderBok.java..........................................................................................94

Hashtag計數器spout.................................................................................................................96

編碼：HashtagCounterBolt.java.........................................................................................96

提交拓撲....................................................................99

編碼：TwitterHashtagStorm.java.......................................................................................99

構建和運行應用程序........................................................101

輸出...................................................................101

在雅虎財經上的應用............................................................102

Spout創(chuàng)建.................................................................103

編碼：YahooFinanceSpout.java........................................................................................103

Bolt倉ij建...................................................................106

編碼：PriceCutOffBolt.java...............................................................................................106

提交拓撲...................................................................109

編碼：YahooFinanceStorm.java........................................................................................109

構建和運行應用程序........................................................110

輸出...................................................................111

應用程序.......................................................................111

Klout.............................................................................................................................................Ill

天氣頻道...................................................................111

電信業(yè).....................................................................112

簡介

什么是ApacheStorm?

ApacheStorm是Twitter開源的分布式實時大數據處理框架，最早開源于

github,從0.9.1版本之后，歸于Apache社區(qū)，被業(yè)界稱為實時版Hadoop。

Storm設計用于在容錯和水平可擴展方法中處理大量數據。它是一個流數據框

架,具有最高的攝取率。雖然Storm是無狀態(tài)的，它通過ApacheZooKeeper管

理分布式環(huán)境和集群狀態(tài)。它很簡單，您可以并行地對實時數據執(zhí)行各種操

作。

隨著越來越多的場景對Hadoop的MapReduce高延遲無法容忍，比如網站統(tǒng)

計、推薦系統(tǒng)、預警系統(tǒng)、金融系統(tǒng)（高頻交易、股票）等等，大數據實時處理

解決方案（流計算）的應用E趨廣泛，目前已是分布式技術領域最新爆發(fā)點，

而Storm更是流計算技術中的佼佼者和主流。

ApacheStorm繼續(xù)成為實時數據分析的領導者。Storm易于設置和操作，并且

它保證每個消息將通過拓撲至少處理一次。

ApacheStormvsHadoop

基本上Hadoop和Storm框架用于分析大數據。兩者互補，在某些方面有所不

同。ApacheSiorm執(zhí)行除持久性之外的所有操作，而Hadoop在所有方面都很

好，但滯后于實時計算。下表比較了Stoi*m和Hadoop的屬在。

StormHadoop

實時流處理批量處理

無狀態(tài)有狀態(tài)

主/從架構與基于ZooKeeper的協(xié)調。主節(jié)點稱為具有/不具有基于ZooKeeper的協(xié)調

nimbus,從屬節(jié)點是主管。的主-從結構。主節(jié)點是作業(yè)跟蹤

器，從節(jié)點是任務跟蹤器，

Storm流過程在集群上每秒可以訪問數萬條消息。Hadoop分布式文件系統(tǒng)（HDFS）使

用MapReduce框架來處理大量的數

據，需要幾分鐘或幾小時。

Storm拓撲運行直到用戶關閉或意外的不可恢復故MapReduce作業(yè)按順序執(zhí)行并最終

障。完成。

兩者都是分布式和容錯的

如果nimbus/supervisor死機，重新啟動使它從它停如果JobTracker死機,所有正在運行

止的地方繼續(xù)，因此沒有什么受到影響。的作業(yè)都會丟失。

使用ApacheStorm的例子

ApacheStorm對于實時大數據流處理非常有名。因此，大多數公司都將Storm

用作其系統(tǒng)的一個組成部分。一些值得注意的例子如下?

Twitter-Twitter正在使用ApacheStorm作為其“發(fā)布商分析產品”。“發(fā)布

商分析產品”處理Twiiter平臺中的每個tweets和點擊。ApacheStorm與

Twitter基礎架構深度集成。

NaviSite-NaviSite正在使用Storm進行事件日志監(jiān)控/審計系統(tǒng)。系統(tǒng)中生成

的每個日志都將通過Slorm。Storm將根據配置的正則表達式集檢查消息，如果

存在匹配，那么該特定消息將保存到數據庫。

Wego-Wego是位于新加坡的旅行元搜索引擎。旅行相關數據來自世界各地的

許多來源，時間不同。Storm幫助Wego搜索實時數據，解決并發(fā)問題，并為最

終用戶找到最佳匹配。

ApacheStorm優(yōu)勢

下面是ApacheStorm提供的好處列表：

?Storm是開源的，強大的，用戶友好的。它可以用于小公司和大公司。

?Storm是容錯的，靈活的，可靠的，并且支持任何編程語言。

?允許實時流處理。

?Storm是令人難以置信的快，因為它具有巨大的處理數據的力量。

?Slorm可以通過線性增加資源來保持性能，即使在負載增加的情況下。它是高

度可擴展的。

?Storm在幾秒鐘或幾分鐘內執(zhí)行數據刷新和端到端傳送響應取決于問題。它具

有非常低的延遲。

?Storm有操作智能。

?Storm提供保證的數據處理，即使群集中的任何連接的節(jié)點死或消息丟失。

Storm的核心組件

?Nimbus:即Storm的Master,負責資源分配和任務調度。一個Storm集群只有一

個

Nimbuso

?Supervisor:即Storm的Slave,負責接收Nimbus分配的任務，管理所有Worke

r,一個Supervisor節(jié)點中包含多個Worker進程。

?Worker:工作進程，每個工作進程中都有多個Task.

■Task:任務，在Storm集群中每個Spout和Bolt都由若干個任務（tasks）來執(zhí)

行。每個任務都與一個執(zhí)行線程相對應。

?Topology:計算拓撲，Storm的拓撲是對實時計算應用邏輯的封裝，它的作用與

MapReduce的任務（Job）很相似，區(qū)別在于MapReduce的一個Job在得到結

果之后總會結束，而拓撲會一直在集群中運行，直到你手動去終止它。拓撲還可以理

解成由一系列通過數據流（StreamGrouping）相互關聯(lián)的Spout和Bolt組成的

的拓撲結構。

?Stream:數據流（Streams）是Storm中最核心的抽象概念。一個數據流指的是在

分布式環(huán)境中并行創(chuàng)建、處理的一組元組（tuple）的無界序列。數據流可以由一種

能夠表述數據流中元組的域（fields）的模式來定義。

?Spout:數據源（Spout）是拓撲中數據流的來源。一般Spout會從一個外部的數

據源讀取元組然后將他們發(fā)送到拓撲中。根據需求的不同，Spout既可以定義為可靠

的數據源，也可以定義為不可靠的數據源。一個可靠的Spout能夠在它發(fā)送的元組

處理失敗時重新發(fā)送該元組，以確保所有的元組都能得到正確的處理；相對應的，不

可靠的Spout就不會在元組發(fā)送之后對元組進行任何其他的處理。一個Spout可

以發(fā)送多個數據流。

?Bolt:拓撲中所有的數據處理均是由Bolt完成的。通過數據過濾（filtering\函數

處理（functions\聚合（aggregations\聯(lián)結（joins\數據庫交互等功能,Bolt

幾乎能夠完成任］可一種數據處理需求。一個Bolt可以實現(xiàn)簡鑿的數據流轉換，而更

復雜的數據流變換通常需要使用多個Bolt并通過多個步驟完成。

?Streamgrouping:為拓撲中的每個Bolt的確定輸入數據流是定義一個拓撲的重

要環(huán)節(jié)。數據流分組定義了在Bolt的不同任務(tasks)中劃分數據流的方式。在S

torm中有八種內置的數據流分組方式。

?Reliability:可靠性。Storm可以通過拓撲來確保每個發(fā)送的元組都能得到正確處

理。通過跟蹤由Spout發(fā)出的每個元組構成的元組樹可以確定元組是否已經完成處

理。每個拓撲都有一個"消息延時"參數，如果Storm在延時時間內沒有檢測到元

組是否處理完成，就會將該元組標記為處理失敗，并會在稍后重新發(fā)送該元組。

Storm程序再Storm集群中運行的示例圖如下：

Topology

為什么把Topology單獨提出來呢,因為Topology是我們開發(fā)程序主要的用的組件。

Topology和MapReduce很相像。

M叩Reduce是M叩進行獲取數據，Reduce進行處理數據。

而Topology則是使用Spout獲取數據，Bolt來進行計算。

總的來說就是一個Topology由一個或者多個的Spout和Bolt組成

具體流程是怎么走，可以通過查看下面這張圖來進行了解。

示例圖：

Stream

^Stream2

.----Stream!

Stream"

0一-一

'、~Stream------Stream

Stream

g

stream—

注:圖片來源/api/tutorials/storm/52o

圖片有三種模式,解釋如下：

第一種比較簡單，就是由一個Spout獲取物g,然后交給一個Bolt進行處理；

第二種稍微復雜點，由一個Spout獲取數據，然后交給一個Bolt進行處理一部分，然后

在交給下一個Bolt進行處理其他部分。

第三種則上瞰復雜，一個Spout可以同時發(fā)送數據到多個Bolt,而一個Bolt也可以接受

多個Spout或多個Bolt,最終形成多個數據流。但是這種數據流必須是有方向的，有起點

和終點，不然會造成死循環(huán)，數據永遠也處理不完。就是Spout發(fā)給Boltl,Boltl發(fā)給

Bolt2,Bolt2又發(fā)給了Boltl,最終形成了一環(huán)狀。

Storm集群安裝

之前已經寫過了，這里就不在說明了。

博客itetlhhttp:〃/2018/01/26/pancm70/

StormHelloWorld

前面講了一些Storm概念，可能在理解上不太清楚，那么這里我們就用一個HelloWorld

代碼示例來體驗下Storm運作的流程吧。

環(huán)境準備

在進行代碼開發(fā)之前，首先得做好相關的準備。

本項目是使用Maven構建的，使用Storm的版本為1.1.1。

Maven的相關依賴如下：

<!一-storm相關jar—>

<dependency>

<groupld>org.apache.storm</groupld>

<artifactId>storm-core</artifactId>

<version>l.1.l</version>

<scope>provided</scope>

</dependency>

具體流程

在寫代碼的時候，我們先來明確要用Storm做什么。

那么第一個程序，就簡單的輸出下信息。

具體步驟如下：

1.啟動topology,設置好Spout和Bolt.

2.將Spout獲取的數據傳遞給Bolt,

3.Bolt接受Spout的數據進行打印。

Spout

那么首先開始編寫Spout類。一股是實現(xiàn)IRichSpout或繼承BaseRichSpout該類,然

后實現(xiàn)該方法。

這里我們繼承BaseRichSpout這個類，該類需要實現(xiàn)這幾個主要的方法：

一、open

open()方法中是在ISpout接口中定義，在Spout組件初始化時被調用。

有三個參數，它們的作用分別是:

1.Storm配置的Map;

2.topology中組件的信息；

3.發(fā)射tuple的方法；

代碼示例：

?Override

publicvoidopen(Mapmap,TopologyContextargl,

SpoutOutputCollectorcollector){

System,out.printlnC^open:imap.get("test"));

this,collector=collector;

)

二、nextTuple

nextTuple。方法是Spout實現(xiàn)的核心。

也就是主要執(zhí)行方法，用于輸出信息，通過collector,emit方法發(fā)射。

這里我們的數據信息已經寫死了，所以這里我們就直接將數據進行發(fā)送。

這里設置只發(fā)送兩次。

代碼示例：

@Override

publicvoidnextTuple(){

if(count<=2){

System.out.printin(〃第〃+count+”次開始發(fā)送數

據???〃)；

this,collector,emit(newValues(message));

count++;

三、declareOutputFields

declareOutputFields是在IComponent接口中定義，用于聲明數據格式。

即輸出的一個Tuple中，包含幾個字段。

因為這里我們只發(fā)射一個,所以就指定一個。如果是多個，則用逗號隔開。

代碼示例：

?Override

publicvoiddeclareOutputFields(OutputFieldsDeclarer

declarer){

System.out.printin("定義格式...;

declarer,declare(newFields(field));

)

四、ack

ack是在ISpout接口中定義，用于表示Tuple處理成功。

代碼示例：

?Override

publicvoidack(Objectobj){

System.out.println("ack:"+obj);

五、fail

fail是在ISpout接口中定義，用于表示Tuple處理失敗。

代碼示例：

@Override

publicvoidfail(Objectobj){

System,out.printin("失敗:〃+obj);

)

六、close

dose是在ISpout接口中定義，用于表示Topology停止。

代碼示例：

?Override

publicvoidclose(){

System,out.printin("關閉...；

)

至于還有其他的，這里就不在一列舉了。

Bolt

Bolt是用于處理數據的組件，主要是由execute方法來進行實現(xiàn)。一般來說需要實

現(xiàn)IRichBolt或繼承BaseRichBolt該類，然后實現(xiàn)其方法。

需要實現(xiàn)方法如下:

一、prepare

在Bolt啟動前執(zhí)行，提供Bolt啟動環(huán)境配置的入口。

參數基本和Sqout-Wo

一般對于不可序列化的對象進行實例化。

這里的我們就簡單的打印下

?Override

publicvoidprepare(Mapmap,TopologyContextargl,

OutputCollectorcollector){

Systpm.out.print1n(^prpparp:zz+map.gpt(，，tpst，/));

this,collector=collector;

)

注:如果是可以序列化的對象,那么最好是使用構造函數。

二、execute

execute。方法是Bolt實現(xiàn)的核心。

也就是執(zhí)行方法，每次Bolt從流凌收f丁閱的tuple,都會調用這個方法。

從tuple中獲取消息可以使用tuple.getStringO和tuple.getStringByFieldO;這

兩個方法。個人推薦第二種，可以通過field來指定接收的消息。

注:如果繼承的是IRichBolt,則需要手動ack。這里就不用了,BaseRichBolt會自動幫我

們應答。

代碼示例：

?Override

publicvoidexecute(Tupletuple){

//Stringmsg=tuple.getString(O);

Stringmsg=tuple.getStringByField("test");

〃這里我們就不做消息的處理，只打印

System.out.printin(，zBolt第〃+count+”接受的消息："+msg);

count++;

/**

*

*沒次調用處理一個輸入的tuple,所有的tuple都必須在一定時間

內應答。

*可以是ack或者fail。否則，spout就會重發(fā)tuple。

*/

//collector,ack(tuple);

)

三、declareOutputFields

和Spout的一樣。

因為到了這里就不再輸出了，所以就什么都沒寫。

?Override

publicvoiddeclareOutputFields(OutputFieldsDeclarerargO)

(

)

cleanup

cleanup是IBolt接口中定義，用于釋放bolt占用的資源。

Storm在終止一個bolt之前會調用這個方法。

因為這里沒有什么資源需要釋放，所以就簡單的打印一句就行了。

?Override

publicvoidcleanup(){

System.out.printin("資源釋放〃)；

)

Topology

這里我們就是用main方法進行提交topology。

不過在提交topology之前，需要進行相應的設置。

這里我就不一細說了，代碼的注釋已經很詳細了。

代碼示例：

importorg.apache.storm.Config;

importorg.apache,storm.LocalCluster;

importorg.apache,storm.StormSubmitter;

importorg.apache,storm,topology.TopologyBuilder;

/**

*

*Title:App

*Description:

*storm測試

*Version:1.0.0

*@authorpancm

*@date2018年3月6日

*/

publicclassApp{

privatestaticfinalStringstrl="testl”;

privatestaticfinalStringstr2="test2〃;

publicstaticvoidmain(String[]args)(

//TODOAuto-generatedmethodstub

〃定義一個拓撲

TopologyBuilderbuilder=newTopologyBuilder();

〃設置一個Executeor(線程),默認一個

builder.setSpout(strl,newTestSpout());

〃設置一個Executeor(線程)，和一個task

builder.setBolt(str2,new

TestBolt(),1).setNumTasks(l).shuffleGrouping(strl);

Configconf=newConfigO;

conf,put("test","test");

try(

〃運行拓撲

if(args!=null&&args.length>0){//有參數時,表示向集群提交

作業(yè)，并把第一個參數當做topology名稱

System.out.printin(〃遠程模式〃)；

StormSubmitter.submitTopology(args[0],conf,

builder.createTopology());

}else{〃沒有參數時，本地提交

〃啟動本地模式

System,out.printIn(〃本地模式〃)；

LocalClustercluster=newLocalCluster();

cluster,submitTopology11177,conf,

builder.createTopology());

Thread,sleep(10000);

//關閉本地集群

cluster,shutdown();

)

}catch(Exceptione)(

e.printStackTraceO;

)

)

}

運行該方法，輸出結果如下：

本地模式

定義格式.??

open:test

第1次開始發(fā)送數據...

第2次開始發(fā)送數據...

prepare:test

Bolt第1接受的消息:這是個測試消息！

Bolt第2接受的消息:這是個測試消息！

資源釋放

關閉...

到這里,是不是基本上對Storm的運作有些了解了呢。

這個demo達到了上述的三種模式圖中的第一種,一個Spout傳輸數據，一個Bolt處理

雌。

那么如果我們想達到第二種模式呢，那又該如何做呢？

假如我們想統(tǒng)計下在一段文本中的單詞出現(xiàn)頻率的話，我們只需執(zhí)行一下步驟就可以了。

.首先將中的消息進行更改為數組,并依次將消息發(fā)送到

1SpoutmessageTestBolto

2.然后TestBolt將獲取的數據進行分割，將分割的數據發(fā)送到TestBolt2。

3,TestBolt2對數據進行統(tǒng)計,在程序關閉的時候進行打印。

4.Topology成功配置并且啟動之后，等待20秒左右，關閉程序，然后得到輸出的結果。

代碼示例如下：

Spout

用于發(fā)送消息。

importjava.util.Map;

importorg.apache,storm,spout.SpoutOutputCollector;

importorg.apache,storm,task.TopologyContext;

importorg.apache,storm,topology.OutputFie1dsDec1arer;

importorg.apache,storm,topology,base.BaseRichSpout;

importorg.apache,storm,tuple.Fields;

importorg.apache,storm,tuple.Values;

/**

*

*Title:TestSpout

*Description:

*發(fā)送信息

*Version:1.0.0

*?authorpancm

*?date2018年3月6日

*/

publicclassTestSpoutextendsBaseRichSpout{

privatestaticfinallongserialVersionUID=

225243592780939490L;

privateSpoutOutputCollectorcollector;

privatestaticfinalStringfield二〃word”;

privateintcount=l;

privateStringf]message={

“Mynicknameisxuwujing",

“Myblogaddressishttp://www.panchengming.com//7,

“Myinterestisplayinggames”

}；

/**

*open。方法中是在ISpout接口中定義，在Spout組件初始化時被調

用。

*有三個參數：

*1.Storm配置的Map;

*2.topology中組件的信息；

*3.發(fā)射tuple的方法;

*/

?Override

publicvoidopen(Mapmap,TopologyContextargl,

SpoutOutputCollectorcollector){

System,out.printIn(^open:z，+map.get("test"));

this,collector=collector;

)

/**

*ncxtTupleO方法是Spout實現(xiàn)的核心。

*也就是主要執(zhí)行方法，用于輸出信息，通過collector.emit方法發(fā)射。

*/

?Override

publicvoidnextTupleO{

if(count<=message.length){

System,out.printin("第〃+count+”次開始發(fā)送數

據…〃)；

this,collector,emit(newValues(message[count-

1]))；

count++;

/**

*declarcOutputFields是在[Component接口中定義，用于聲明數據格

式。

*即輸出的一個Tuple中，包含幾個字段。