基于案例Storm實(shí)時(shí)流計(jì)算

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

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

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

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

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

activate和deactivate

：一個(gè)spout可以被暫時(shí)激活和關(guān)閉，這兩個(gè)方法分別在對(duì)應(yīng)的時(shí)刻被調(diào)用。

nextTuple

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

ack(Object)

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

fail(Object)

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

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

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

IBolt繼承了java.io.Serializable，我們?cè)趎imbus上提交了topology以后，創(chuàng)建出來的bolt會(huì)序列化后發(fā)送到具體執(zhí)行的worker上去。worker在執(zhí)行該Bolt時(shí)，會(huì)先調(diào)用prepare方法傳入當(dāng)前執(zhí)行的上下文execute接受一個(gè)tuple進(jìn)行處理，并用prepare方法傳入的OutputCollector的ack方法（表示成功）或fail（表示失?。﹣矸答佁幚斫Y(jié)果cleanup同ISpout的close方法，在關(guān)閉前調(diào)用。同樣不保證其一定執(zhí)行execute最重要的方法，用來處理自己的業(yè)務(wù)邏輯declareOutputFields

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