基于案例Storm實時流計算

基于案例講解Storm實時計算目錄Storm的應用場景舉例Storm簡介Storm的主要特點Storm組件Storm編程模型Storm安裝Storm實例講解Storm的應用場景舉例淘寶雙十一實時銷售額統(tǒng)計車輛7*24小時監(jiān)控電信行業(yè)重大節(jié)假日實時保障監(jiān)控電商實時推薦1.

Storm是一個分布式的、容錯的實時計算系統(tǒng),它采用Clojure編寫的2.Storm可被用于“流處理”之中，實時處理消息并更新數(shù)據(jù)庫3.Storm可以進行連續(xù)查詢并把結果即時反饋給客戶，比如將Twitter上的熱門話題發(fā)送到客戶端4.Storm可以用來并行處理密集查詢，Storm的拓撲結構是一個等待調用信息的分布函數(shù)，當它收到一條調用信息后，會對查詢進行計算，并返回查詢結果。Storm簡介Storm的主要特點1.簡單的編程模型。類似于MapReduce降低了并行批處理復雜性，Storm降低了進行實時處理的復雜性。2.可以使用各種編程語言。你可以在Storm之上使用各種編程語言。默認支持Clojure、Java、Ruby和Python。3.水平擴展。計算是在多個線程、進程和服務器之間并行進行的。4.可靠的消息處理。Storm保證每個消息至少能得到一次完整處理。任務失敗時，它會負責從消息源重試消息。5.快速。系統(tǒng)的設計保證了消息能得到快速的處理，使用ZeroMQ作為其底層消息隊列。6.本地模式。Storm有一個“本地模式”，可以在處理過程中完全模擬Storm集群。這讓你可以快速進行開發(fā)和單元測試。Storm組件首先我們通過一個storm和hadoop的對比來了解storm中的基本概念。hadoopstorm系統(tǒng)角色JobtrackerNimbusTasktrackerSupervisorChildWorker應用名稱JobTopology組件接口M/RSpout/Bolt運行狀態(tài)數(shù)據(jù)處理完就結束一旦運行，根本停不下來。。。Nimbus：負責資源分配和任務調度。Supervisor：負責接受nimbus分配的任務，啟動和停止屬于自己管理的worker進程。Worker：運行具體處理組件邏輯的進程。Task：worker中每一個spout/bolt的線程稱為一個task.在storm0.8之后，task不再與物理線程對應，同一個spout/bolt的task可能會共享一個物理線程，該線程稱為executor。Storm組件下面這個圖描述了以上幾個角色之間的關系。NimbuszookeeperzookeeperzookeeperSupervisorSupervisorSupervisorSupervisorSupervisorSupervisorWorkerWorkerWorker主控節(jié)點，用于提交任務，分配集群任務，監(jiān)控集群狀態(tài)協(xié)調，存放集群的公共數(shù)據(jù)(心跳，集群狀態(tài)，配置信息),Nimbus分配給Supervisor的任務接受Nimbus分配的任務，管理自己的Worker進程具體的處理邏輯組件Storm組件在storm中，應用程序的實現(xiàn)實時處理的邏輯被封裝在Topology中一個Topology是由Spout組件(數(shù)據(jù)源)和Bolt組件(數(shù)據(jù)操作)通過StreamGroupings進行連接的TopologySpout數(shù)據(jù)源是否在高速上在高速上，速度是否>120<=120>120存入超速數(shù)據(jù)庫不在高速上，速度是否>80>80<=80實時監(jiān)控超速車輛Storm組件簡而言之，Spout從來源處讀取數(shù)據(jù)并放入topology。Spout分成可靠和不可靠兩種；當Storm接收失敗時，可靠的Spout會對tuple（元組，數(shù)據(jù)項組成的列表）進行重發(fā)；而不可靠的Spout不會考慮接收成功與否只發(fā)射一次。而Spout中最主要的方法就是nextTuple（），該方法會發(fā)射一個新的tuple到topology，如果沒有新tuple發(fā)射則會簡單的返回一個Topology是由Spout組件(數(shù)據(jù)源)和Bolt組件(數(shù)據(jù)操作)通過StreamTopology中所有的處理都由Bolt完成。Bolt可以完成任何事，比如：連接的過濾、聚合、訪問文件/數(shù)據(jù)庫、等等。Bolt從Spout中接收數(shù)據(jù)并進行處理，如果遇到復雜流的處理也可能將tuple發(fā)送給另一個Bolt進行處理。而Bolt中最重要的方法是execute（），以新的tuple作為參數(shù)接收。不管是Spout還是Bolt，如果將tuple發(fā)射成多個流，這些流都可以通過declareStream（）來聲明。SpoutBoltStorm組件StreamGrouping定義了一個流在Bolt任務間該如何被切分。隨機分組（Shufflegrouping）：隨機分發(fā)tuple到Bolt的任務，保證每個任務獲得相等數(shù)量的tuple。字段分組（Fieldsgrouping）：根據(jù)指定字段分割數(shù)據(jù)流，并分組。例如，根據(jù)“user-id”字段，相同“user-id”的元組總是分發(fā)到同一個任務，不同“user-id”的元組可能分發(fā)到不同的任務。全部分組（Allgrouping）：tuple被復制到bolt的所有任務。這種類型需要謹慎使用。全局分組（Globalgrouping）：全部流都分配到bolt的同一個任務。明確地說，是分配給ID最小的那個task。無分組（Nonegrouping）：你不需要關心流是如何分組。目前，無分組等效于隨機分組。但最終，Storm將把無分組的Bolts放到Bolts或Spouts訂閱它們的同一線程去執(zhí)行（如果可能）。直接分組（Directgrouping）：這是一個特別的分組類型。元組生產者決定tuple由哪個元組處理者任務接收。當然還可以實現(xiàn)CustomStreamGroupimg接口來定制自己需要的分組。StreamGroupingStorm編程模型

可實現(xiàn)接口Ispout,或繼承BaseSpout主要方法：

open方法是初始化動作。允許你在該spout初始化時做一些動作，傳入了上下文，方便取上下文的一些數(shù)據(jù)。

close方法在該spout關閉前執(zhí)行，但是并不能得到保證其一定被執(zhí)行。spout是作為task運行在worker內，在cluster模式下，supervisor會直接kill-9woker的進程，這樣它就無法執(zhí)行了。而在本地模式下，只要不是kill-9,如果是發(fā)送停止命令，是可以保證close的執(zhí)行的。

activate和deactivate

：一個spout可以被暫時激活和關閉，這兩個方法分別在對應的時刻被調用。

nextTuple

用來發(fā)射數(shù)據(jù)。

ack(Object)

傳入的Object其實是一個id，唯一表示一個tuple。該方法是這個id所對應的tuple被成功處理后執(zhí)行。

fail(Object)

同ack，只不過是tuple處理失敗時執(zhí)行。declareOutputFields

申明要發(fā)射的字段SpoutStorm編程模型

可實現(xiàn)接口IBolt,或BaseRichbolt主要方法：

IBolt繼承了java.io.Serializable，我們在nimbus上提交了topology以后，創(chuàng)建出來的bolt會序列化后發(fā)送到具體執(zhí)行的worker上去。worker在執(zhí)行該Bolt時，會先調用prepare方法傳入當前執(zhí)行的上下文execute接受一個tuple進行處理，并用prepare方法傳入的OutputCollector的ack方法（表示成功）或fail（表示失敗）來反饋處理結果cleanup同ISpout的close方法，在關閉前調用。同樣不保證其一定執(zhí)行execute最重要的方法，用來處理自己的業(yè)務邏輯declareOutputFields

申明要發(fā)射的字段BoltStorm安裝搭建Zookeeper集群安裝Storm依賴庫(ZeromQ,JZMQ,JDK,Python)下載并解壓Storm發(fā)布版本修改storm.yaml配置文件啟動Storm各個后臺進程Nimbus：bin/stormnimbusSupervisor:bin/stormsupervisorUI:stormuiStorm實例講解仿淘寶雙十一實時監(jiān)控銷售額1.從activemq中接受訂單信息(數(shù)據(jù)格式:用戶id|時間|金額|商品id|商家id)2.計算訂單金額，一分鐘輸出一次3.合并計算結果，并寫入mysqlmqSpout從mq中接受訂單信息CountBolt計算金額MergeBolt匯總，寫入Mysql前臺界面展示，每30秒查一次數(shù)據(jù)庫CountBolt計算金額CountBolt計算金額CountBolt計算金額Storm實例講解仿淘寶雙十一實時監(jiān)控銷售額1.從activemq中接受訂單信息(數(shù)據(jù)格式:用戶id|時間|金額|商品id|商家id)