storm研究與介紹課件

研究與介紹Storm目錄2、Storm的組件1、Storm是什么？4、項(xiàng)目架構(gòu)3、Storm應(yīng)用場(chǎng)景1

.

Storm是什么？Master對(duì)Map的預(yù)處理：對(duì)于大量的數(shù)據(jù)進(jìn)行切分，

劃分為M個(gè)16~64M的數(shù)據(jù)分片(可通過參數(shù)自定義分片大小)調(diào)用Mapper函數(shù)：Master為Worker分配Map任務(wù)，

每個(gè)分片都對(duì)應(yīng)一個(gè)Worker進(jìn)行處理。各個(gè)Worker讀取

并調(diào)用用戶定義的Mapper函數(shù)處理數(shù)據(jù)，并將結(jié)果存入

HDFS，返回存儲(chǔ)位置給Master。

一個(gè)Worker在Map階段完成時(shí)，在HDFS中，

生成一個(gè)排好序的Key-values組成的文件。并將位置信息匯報(bào)給Master。

先來聊聊我們熟悉的Hadoop全量數(shù)據(jù)處理使用的大多是鼎鼎大名的hadoop或者h(yuǎn)ive，作為一個(gè)批處理系統(tǒng)，hadoop以其吞吐量大、自動(dòng)容錯(cuò)等優(yōu)點(diǎn)，在海量數(shù)據(jù)處理上得到了廣泛的使用。Hadoop下的Map/Reduce框架對(duì)于數(shù)據(jù)的處理流程是：1、將要處理的數(shù)據(jù)上傳到Hadoop的文件系統(tǒng)HDFS中。2、Map階段以上的流程，粗略概括，就是從HDFS中獲取數(shù)據(jù)，將其按照大小分片，進(jìn)行分布式處理，最終輸出結(jié)果。從流程來看，Hadoop框架進(jìn)行數(shù)據(jù)處理有以下要求：1、數(shù)據(jù)已經(jīng)存在在HDFS當(dāng)中。2、數(shù)據(jù)間是少關(guān)聯(lián)的。各個(gè)任務(wù)執(zhí)行器在執(zhí)行負(fù)責(zé)的數(shù)據(jù)時(shí)，無需考慮對(duì)其他數(shù)據(jù)

的影響，數(shù)據(jù)之間應(yīng)盡可能是無聯(lián)系、不會(huì)影響的。使用Hadoop，適合大批量

的數(shù)據(jù)處理，這是他所擅長(zhǎng)的。由于基于Map/Reduce這種單級(jí)的

數(shù)據(jù)處理模型進(jìn)行，因此，如果數(shù)據(jù)間的關(guān)聯(lián)系較大，

需要進(jìn)行數(shù)據(jù)的多級(jí)交互處理（某個(gè)階段的處理數(shù)據(jù)依賴于上

一個(gè)階段），需要進(jìn)行多次map/reduce。又由于map/reduce

每次執(zhí)行都需要遍歷整個(gè)數(shù)據(jù)集，對(duì)于數(shù)據(jù)的實(shí)時(shí)計(jì)算

并不合適，于是有了storm3、Reduce階段a)Master對(duì)Reduce的預(yù)處理：Master為Worker分配Reduce任務(wù)，他會(huì)將所有Mapper產(chǎn)生的數(shù)據(jù)進(jìn)行映射，將相同key的任務(wù)分配給某個(gè)Worker。b)調(diào)用Reduce函數(shù)：各個(gè)Worker將分配到的數(shù)據(jù)集進(jìn)行排序（使用工具類Merg），并調(diào)用用戶自定義的Reduce函數(shù)，并將結(jié)果寫入HDFS。每個(gè)Worker的Reduce任務(wù)完成后，都會(huì)在HDFS中生成一個(gè)輸出文件。Hadoop并不將這些文件合并，因?yàn)檫@些文件往往會(huì)作為另一個(gè)Map/reduce程序的輸入。那么問題來了，What

is

Storm?首先我們打開官網(wǎng)，可以看到：2.Storm的組件Storm集群和Hadoop集群表面上看很類似。但是Hadoop上運(yùn)行的是MapReducejobs，而在Storm上運(yùn)行的是拓?fù)洌╰opology），這兩者之間是非常不一樣的。一個(gè)關(guān)鍵的區(qū)別是：一個(gè)MapReducejob最終會(huì)結(jié)束，而一個(gè)topology永遠(yuǎn)會(huì)運(yùn)行（除非你手動(dòng)kill掉）。Storm集群主要由一個(gè)主節(jié)點(diǎn)（Nimbus后臺(tái)程序）和一群工作節(jié)點(diǎn)（workernode）Supervisor的節(jié)點(diǎn)組成，通過Zookeeper進(jìn)行協(xié)調(diào)。Nimbus類似Hadoop里面的JobTracker。Nimbus負(fù)責(zé)在集群里面分發(fā)代碼，分配計(jì)算任務(wù)給機(jī)器，并且監(jiān)控狀態(tài)。每一個(gè)工作節(jié)點(diǎn)上面運(yùn)行一個(gè)叫做Supervisor的節(jié)點(diǎn)。Supervisor會(huì)監(jiān)聽分配給它那臺(tái)機(jī)器的工作，根據(jù)需要啟動(dòng)/關(guān)閉工作進(jìn)程。每一個(gè)工作進(jìn)程執(zhí)行一個(gè)topology的一個(gè)子集；一個(gè)運(yùn)行的topology由運(yùn)行在很多機(jī)器上的很多工作進(jìn)程組成。Topology示意圖：1、Nimbus主節(jié)點(diǎn)：主節(jié)點(diǎn)通常運(yùn)行一個(gè)后臺(tái)程序——Nimbus，用于響應(yīng)分布在集群中的節(jié)點(diǎn)，分配任務(wù)和監(jiān)測(cè)故障。這個(gè)很類似于Hadoop中的JobTracker。2、Supervisor工作節(jié)點(diǎn)：工作節(jié)點(diǎn)同樣會(huì)運(yùn)行一個(gè)后臺(tái)程序——Supervisor，用于收聽工作指派并基于要求運(yùn)行工作進(jìn)程。每個(gè)工作節(jié)點(diǎn)都是topology中一個(gè)子集的實(shí)現(xiàn)。而Nimbus和Supervisor之間的協(xié)調(diào)則通過Zookeeper系統(tǒng)或者集群。3、Zookeeper

Zookeeper是完成Supervisor和Nimbus之間協(xié)調(diào)的服務(wù)。而應(yīng)用程序?qū)崿F(xiàn)實(shí)時(shí)的邏輯則被封裝進(jìn)Storm中的“topology”。topology則是一組由Spouts（數(shù)據(jù)源）和Bolts（數(shù)據(jù)操作）通過StreamGroupings進(jìn)行連接的圖。下面對(duì)出現(xiàn)的術(shù)語進(jìn)行更深刻的解析。4、Worker：運(yùn)行具體處理組件邏輯的進(jìn)程。5、Task：

worker中每一個(gè)spout/bolt的線程稱為一個(gè)task.在storm0.8之后，task不再與物理線程對(duì)應(yīng)，同一個(gè)spout/bolt的task可能會(huì)共享一個(gè)物理線程，該線程稱為executor。6、Topology（拓?fù)洌?/p>

storm中運(yùn)行的一個(gè)實(shí)時(shí)應(yīng)用程序，因?yàn)楦鱾€(gè)組件間的消息流動(dòng)形成邏輯上的一個(gè)拓?fù)浣Y(jié)構(gòu)。一個(gè)topology是spouts和bolts組成的圖，通過streamgroupings將圖中的spouts和bolts連接起來，如下圖：一個(gè)topology會(huì)一直運(yùn)行直到你手動(dòng)kill掉，Storm自動(dòng)重新分配執(zhí)行失敗的任務(wù)，并且Storm可以保證你不會(huì)有數(shù)據(jù)丟失（如果開啟了高可靠性的話）。如果一些機(jī)器意外停機(jī)它上面的所有任務(wù)會(huì)被轉(zhuǎn)移到其他機(jī)器上。運(yùn)行一個(gè)topology很簡(jiǎn)單。首先，把你所有的代碼以及所依賴的jar打進(jìn)一個(gè)jar包。然后運(yùn)行類似下面的這個(gè)命令：

stormjarall-my-code.jar

backtype.storm.MyTopologyarg1arg2這個(gè)命令會(huì)運(yùn)行主類:backtype.strom.MyTopology,參數(shù)是arg1,arg2。這個(gè)類的main函數(shù)定義這個(gè)topology并且把它提交給Nimbus。stormjar負(fù)責(zé)連接到Nimbus并且上傳jar包。Topology的定義是一個(gè)Thrift結(jié)構(gòu)，并且Nimbus就是一個(gè)Thrift服務(wù)，你可以提交由任何語言創(chuàng)建的topology。上面的方面是用JVM-based語言提交的最簡(jiǎn)單的方法。7、Spout：消息源spout是Storm里面一個(gè)topology里面的消息生產(chǎn)者。簡(jiǎn)而言之，Spout從來源處讀取數(shù)據(jù)并放入topology。Spout分成可靠和不可靠?jī)煞N；當(dāng)Storm接收失敗時(shí)，可靠的Spout會(huì)對(duì)tuple（元組，數(shù)據(jù)項(xiàng)組成的列表）進(jìn)行重發(fā)；而不可靠的Spout不會(huì)考慮接收成功與否只發(fā)射一次。消息源可以發(fā)射多條消息流stream。使用OutputFieldsDeclarer.declareStream來定義多個(gè)stream，然后使用SpoutOutputCollector來發(fā)射指定的stream。而Spout中最主要的方法就是nextTuple（），該方法會(huì)發(fā)射一個(gè)新的tuple到topology，如果沒有新tuple發(fā)射則會(huì)簡(jiǎn)單的返回。要注意的是nextTuple方法不能阻塞，因?yàn)閟torm在同一個(gè)線程上面調(diào)用所有消息源spout的方法。另外兩個(gè)比較重要的spout方法是ack和fail。storm在檢測(cè)到一個(gè)tuple被整個(gè)topology成功處理的時(shí)候調(diào)用ack，否則調(diào)用fail。storm只對(duì)可靠的spout調(diào)用ack和fail。8、Bolt：

Topology中所有的處理都由Bolt完成。即所有的消息處理邏輯被封裝在bolts里面。Bolt可以完成任何事，比如：連接的過濾、聚合、訪問文件/數(shù)據(jù)庫(kù)、等等。

Bolt從Spout中接收數(shù)據(jù)并進(jìn)行處理，如果遇到復(fù)雜流的處理也可能將tuple發(fā)送給另一個(gè)Bolt進(jìn)行處理。即需要經(jīng)過很多blots。比如算出一堆圖片里面被轉(zhuǎn)發(fā)最多的圖片就至少需要兩步：第一步算出每個(gè)圖片的轉(zhuǎn)發(fā)數(shù)量。第二步找出轉(zhuǎn)發(fā)最多的前10個(gè)圖片。（如果要把這個(gè)過程做得更具有擴(kuò)展性那么可能需要更多的步驟）。

Bolts可以發(fā)射多條消息流，使用OutputFieldsDeclarer.declareStream定義stream，使用OutputCollector.emit來選擇要發(fā)射的stream。而Bolt中最重要的方法是execute（），以新的tuple作為參數(shù)接收。不管是Spout還是Bolt，如果將tuple發(fā)射成多個(gè)流，這些流都可以通過declareStream（）來聲明。

bolts使用OutputCollector來發(fā)射tuple，bolts必須要為它處理的每一個(gè)tuple調(diào)用OutputCollector的ack方法，以通知Storm這個(gè)tuple被處理完成了，從而通知這個(gè)tuple的發(fā)射者spouts。一般的流程是：bolts處理一個(gè)輸入tuple,發(fā)射0個(gè)或者多個(gè)tuple,然后調(diào)用ack通知storm自己已經(jīng)處理過這個(gè)tuple了。storm提供了一個(gè)IBasicBolt會(huì)自動(dòng)調(diào)用ack。9、Tuple：一次消息傳遞的基本單元。本來應(yīng)該是一個(gè)key-value的map，但是由于各個(gè)組件間傳遞的tuple的字段名稱已經(jīng)事先定義好，所以tuple中只要按序填入各個(gè)value就行了，所以就是一個(gè)valuelist.10、Stream：源源不斷傳遞的tuple就組成了stream。消息流stream是storm里的關(guān)鍵抽象。一個(gè)消息流是一個(gè)沒有邊界的tuple序列，而這些tuple序列會(huì)以一種分布式的方式并行地創(chuàng)建和處理。通過對(duì)stream中tuple序列中每個(gè)字段命名來定義stream。在默認(rèn)的情況下，tuple的字段類型可以是：integer，long，short，byte，string，double，float，boolean和bytearray。你也可以自定義類型（只要實(shí)現(xiàn)相應(yīng)的序列化器）。每個(gè)消息流在定義的時(shí)候會(huì)被分配給一個(gè)id，因?yàn)閱蜗蛳⒘魇褂玫南喈?dāng)普遍，OutputFieldsDeclarer定義了一些方法讓你可以定義一個(gè)stream而不用指定這個(gè)id。在這種情況下這個(gè)stream會(huì)分配個(gè)值為‘default’默認(rèn)的id。

Storm提供的最基本的處理stream的原語是spout和bolt。你可以實(shí)現(xiàn)spout和bolt提供的接口來處理你的業(yè)務(wù)邏輯。11、StreamGroupings：StreamGrouping定義了一個(gè)流在Bolt任務(wù)間該如何被切分。這里有Storm提供的6個(gè)StreamGrouping類型：1）.隨機(jī)分組（Shufflegrouping）：隨機(jī)分發(fā)tuple到Bolt的任務(wù)，保證每個(gè)任務(wù)獲得相等數(shù)量的tuple。2）.字段分組（Fieldsgrouping）：根據(jù)指定字段分割數(shù)據(jù)流，并分組。例如，根據(jù)“user-id”字段，相同“user-id”的元組總是分發(fā)到同一個(gè)任務(wù)，不同“user-id”的元組可能分發(fā)到不同的任務(wù)。3）.全部分組（Allgrouping）：tuple被復(fù)制到bolt的所有任務(wù)。這種類型需要謹(jǐn)慎使用。4）.全局分組（Globalgrouping）：全部流都分配到bolt的同一個(gè)任務(wù)。明確地說，是分配給ID最小的那個(gè)task。5）.無分組（Nonegrouping）：你不需要關(guān)心流是如何分組。目前，無分組等效于隨機(jī)分組。但最終，Storm將把無分組的Bolts放到Bolts或Spouts訂閱它們的同一線程去執(zhí)行（如果可能）。6）.直接分組（Directgrouping）：這是一個(gè)特別的分組類型。元組生產(chǎn)者決定tuple由哪個(gè)元組處理者任務(wù)接收。當(dāng)然還可以實(shí)現(xiàn)CustomStreamGroupimg接口來定制自己需要的分組。storm和hadoop的對(duì)比來了解storm中的基本概念。HadoopStorm系統(tǒng)角色JobTrackerNimbusTaskTrackerSupervisorChildworker應(yīng)用名稱JobTopology組件接口Mapper/ReducerSpout/Bolt3.Storm應(yīng)用場(chǎng)景

Storm與其他大數(shù)據(jù)解決方案的不同之處在于它的處理方式。Hadoop

在本質(zhì)上是一個(gè)批處理系統(tǒng)。數(shù)據(jù)被引入Hadoop

文件系統(tǒng)(HDFS)并分發(fā)到各個(gè)節(jié)點(diǎn)進(jìn)行處理。當(dāng)處理完成時(shí)，結(jié)果數(shù)據(jù)返回到HDFS供始發(fā)者使用。Storm支持創(chuàng)建拓?fù)浣Y(jié)構(gòu)來轉(zhuǎn)換沒有終點(diǎn)的數(shù)據(jù)流。不同于Hadoop

作業(yè)，這些轉(zhuǎn)換從不停止，它們會(huì)持續(xù)處理到達(dá)的數(shù)據(jù)。Twitter列舉了Storm的三大類應(yīng)用：1.信息流處理{Streamprocessing}Storm可用來實(shí)時(shí)處理新數(shù)據(jù)和更新數(shù)據(jù)庫(kù)，兼具容錯(cuò)性和可擴(kuò)展性。即Storm可以用來處理源源不斷流進(jìn)來的消息，處理之后將結(jié)果寫入到某個(gè)存儲(chǔ)中去。2.連續(xù)計(jì)算{Continuouscomputation}Storm可進(jìn)行連續(xù)查詢并把結(jié)果即時(shí)反饋給客戶端。比如把Twitter上的熱門話題發(fā)送到瀏覽器中。3.分布式遠(yuǎn)程程序調(diào)用{DistributedRPC}Storm可用來并行處理密集查詢。Storm的拓?fù)浣Y(jié)構(gòu)是一個(gè)等待調(diào)用信息的分布函數(shù)，當(dāng)它收到一條調(diào)用信息后，會(huì)對(duì)查詢進(jìn)行計(jì)算，并返回查詢結(jié)果。舉個(gè)例子DistributedRPC可以做并行搜索或者處理大集合的數(shù)據(jù)。通過配置drpc服務(wù)器，將storm的topology發(fā)布為drpc服務(wù)。客戶端程序可以調(diào)用drpc服務(wù)將數(shù)據(jù)發(fā)送到storm集群中，并接收處理結(jié)果的反饋。這種方式需要drpc服務(wù)器進(jìn)行轉(zhuǎn)發(fā)，其中drpc服務(wù)器底層通過thrift實(shí)現(xiàn)。適合的業(yè)務(wù)場(chǎng)景主要是實(shí)時(shí)計(jì)算。并且擴(kuò)展性良好，可以增加每個(gè)節(jié)點(diǎn)的工作worker數(shù)量來動(dòng)態(tài)擴(kuò)展。4.項(xiàng)目架構(gòu)系統(tǒng)需要滿足的要求實(shí)時(shí)的在

任意大小的數(shù)據(jù)集上執(zhí)行各種類型的操作處理架構(gòu)設(shè)計(jì)SpeedLayerServingLayerBatchLayer架構(gòu)設(shè)計(jì)SpeedLayerServingLayerBatchLayerBatchLayerquery=function(alldata)1、每次查詢都在數(shù)據(jù)集上重新執(zhí)行一次2、預(yù)計(jì)算BatchLayer批處理層的主要作用就是進(jìn)行各種復(fù)雜的預(yù)計(jì)算，預(yù)計(jì)算的結(jié)果稱作BatchView。再次查詢時(shí)直接從BatchView中讀取結(jié)果數(shù)據(jù)，為了支持快速的隨機(jī)的查詢需求，需要對(duì)BatchView中的數(shù)據(jù)建立索引。BatchLayerAllDataBatchLayerBatchViewBatchViewBatchViewBatchLayer批處理層的主要職責(zé)包括：1、存儲(chǔ)不變的、持續(xù)增加的海量數(shù)據(jù)集記錄2、預(yù)計(jì)算各種Query操作對(duì)應(yīng)的BatchViewHDFSMapReduceBatchLayer

批處理層將Query的結(jié)果保存到了BatchView中，而Hadoop的MapReduce所對(duì)應(yīng)的BatchView是存儲(chǔ)在HDFS上一系列文件，我們無法對(duì)這些結(jié)果文件進(jìn)行快速的隨機(jī)的訪問。架構(gòu)設(shè)計(jì)SpeedLayerServingLayer

BatchLayerServingLayer

ServingLayer的主要作用就是對(duì)

BatchView建立索引，

以支持業(yè)務(wù)層的快速的隨機(jī)的訪問需求。ServingLayer

分布式數(shù)據(jù)庫(kù)是ServingLayer的最佳選擇。分布式數(shù)據(jù)庫(kù)需要滿足的條件：1、支持BatchView的批量導(dǎo)入2、支持對(duì)BatchView建立索引和隨機(jī)讀取BulkLoad內(nèi)建的索引機(jī)制ServingLayerServingLayer面臨的其他問題：BatchLayer的計(jì)算邏輯每天執(zhí)行一次，所以ServingLayer對(duì)應(yīng)的BatchView每天更新一次，從而導(dǎo)致最新的數(shù)據(jù)無法實(shí)時(shí)的更新到ServingLayer架構(gòu)設(shè)計(jì)SpeedLayerServingLayerBatchLayerSpeedLayerSpeedLayer的主要目的是彌補(bǔ)系統(tǒng)存在的對(duì)ServingLayer延遲更新的缺陷。SpeedLayer會(huì)對(duì)進(jìn)入到系統(tǒng)的最新的數(shù)據(jù)進(jìn)行各種實(shí)時(shí)的處理，并產(chǎn)生

RealTimeView。SpeedLayerNewDataSpeedLayerRealTimeViewRealTimeViewRealTimeViewSpeedLayerSpeedLayer使用分布式的、可靠的、實(shí)時(shí)的計(jì)算系統(tǒng)--------------Storm。

RealTimeView使用支持隨機(jī)讀、隨機(jī)寫的分布式數(shù)據(jù)庫(kù)HBase。SpoutBoltBoltBoltHBasequery=function(alld