storm 1 0版本滑動視窗的實現及原理

原文和作者一起討論:

這裡主要演示在storm1.0以後如何通過繼承storm1.0提供的類來快速開發出視窗滑動的功能。視窗可以從時間或數量上來劃分，由如下兩個因素決定：視窗的長度，可以是時間間隔或tuple數量；滑動間隔(sliding interval)，可以是時間間隔或tuple數量。比如：每兩秒統計最近6秒的請求數量；每接收2個tuple就統計最近接收的6個tuple的平均值......。

storm1.0支援的時間和數量的排列組合有如下：

withwindow(count windowlength, count slidinginterval)

每收到slidinginterval條資料統計最近的windowlength條資料。

withwindow(count windowlength)

每收到1條資料統計最近的windowlength條資料。

withwindow(count windowlength, duration slidinginterval)

每過slidinginterval秒統計最近的windowlength條資料。

withwindow(duration windowlength, count slidinginterval)

每收到slidinginterval條資料統計最近的windowlength秒的資料。

withwindow(duration windowlength, duration slidinginterval)

每過slidinginterval秒統計最近的windowlength秒的資料。

public withwindow(duration windowlength)

每收到1條資料統計最近的windowlength秒的資料。

接下來，簡單的演示如何使用storm1.0實現滑動視窗的功能,先編寫spout類，randomsentencespout

負責傳送乙個整形數值，數值每次傳送都會自動加一，且

randomsentencespout

固定每隔兩秒向

bolt傳送一次資料。

randomsentencespout

和前面關於

spout的講解一樣。

1.public
class randomsentencespout extends baserichspout 12.13.    public
void open(map conf, topologycontext context, 14.                          spoutoutputcollector collector) 17.18.    public
void nexttuple()  catch (interruptedexception e) 27.    }}

滑動視窗的邏輯實現的重點是bolt類，這裡我們編寫slidingwindowbolt

類讓它繼承乙個新的類名為

basewindowedbolt

來獲得視窗計數的功能。

basewindowedbolt

和前面的

base

bolt

和basewindowedbolt

提供的方法名都一樣，只是

execute方法的引數型別為tuplewindow,tuplewindow引數裡面裝載了乙個視窗長度類的tuple資料。通過對tuplewindow遍歷，我們可以計算這乙個視窗內tuple數的平均值或總和等指標。具體見**12-16行，統計了乙個視窗內的數值型資料的總和。

1.public
class slidingwindowbolt extends basewindowedbolt 10.11.    public
void execute(tuplewindow inputwindow) 19.        system.out.println("*****=="+sum);20. //
collector.emit(new values(sum));21.    }22.23.    @override24.    public
void declareoutputfields(outputfieldsdeclarer declarer) }

我們已經實現了視窗計數的邏輯**，現在我們需要提供topology來指明各個元件的關係，以及指定slidingwindowbolt

的視窗的組合，這裡我們演示了如何每兩秒統計最近

6秒的數值總和，如果注釋掉10-13行**，去掉5-8行的注釋，這個topology就是告訴

slidingwindowbolt

每接收到兩條

tuple就統計最近接收到的6條tuple的數值的總和。

1.public
class windowstopology }

這裡演示的是bolt節點併發度為1的視窗功能，實際生產中，因為資料量很大，往往將bolt節點的併發度設定為多個，這個時候我們的slidingwindowbolt

就無法統計出乙個視窗的數值總和了。因為每乙個

bolt的並行節點只能統計自己乙個視窗接收到資料的總和，無法統計出乙個視窗內全域性資料的總和，借助redis來實現是可以的，但是必須引入redis的事務機制或者借助分布式鎖，否則會出現髒資料的情況。在這裡我們介紹另一種實現方式就是靈活的使用storm提供的視窗功能，只是視窗的tuple數。

仍然是使用上面提供的類，只是我們增加乙個bolt類，來統計每個slidingwindowbolt

節點傳送給它的數值。

1.public
class countword extends basewindowedbolt11.    12.    public
void execute(tuplewindow inputwindow) 19.         system.out.println("乙個視窗內的總值為:"+sum);20.    }21.22.    public
void declareoutputfields(outputfieldsdeclarer declarer) }

然後我們注釋

randomsentencespout

第22行**，取消對23行**的注釋，方便觀察結果。去掉

slidingwindowbolt

類20和25行**。

topology啟動類如下：

1.public
class windowstopology }

storm 1 0版本滑動視窗的實現及原理

列轉行1 0版本

低於1 0版本 pod使用

Windows 10 版本資訊

storm 1 0版本滑動視窗的實現及原理

列轉行1 0版本

低於1 0版本 pod使用

Windows 10 版本資訊

相關推薦