《大數據》xxu@許鑫華東師范大學經濟與管理學部E-mail:xxu@/xuxin流計算大數據Big

Data華東師范大學信第息一管頁理，共系七十三頁。

許鑫《大數據》xxu@提綱流計算概述流計算處理流程流計算應用流計算開源框架–StormSpark

StreamingSamzaStorm、Spark

Streaming和Samza的應用場景Storm編程實踐華東師范大學信第息二管頁理，共系七十三頁。

許鑫《大數據》xxu@華東師范大學信第息三管頁理，共系七十三頁。

許鑫8.1

流計算概述?8.1.1靜態數據和流數據?8.1.2批量計算和實時計算?8.1.3流計算概念?8.1.4流計算與Hadoop?8.1.5流計算框架《大數據》xxu@

很多企業為了支持決策分析而構建的數據倉庫系統，其中存放的大量歷史數據就是靜態數據。技術人員可以利用數據挖掘和OLAP（On-Line

AnalyticalProcessing）分析工具從靜態數據中找到對企業有價值的信息8.1.1靜態數據和流數據華東師范大學信第息四管頁理，共系七十三頁。

許鑫《大數據》xxu@華東師范大學信第息五管頁理，共系七十三頁。

許鑫

近年來，在Web應用、網絡監控、傳感監測等領域，興起了一種新的數據密集型應用——流數據，即數據以大量、快速、時變的流形式持續到達實例：PM2.5檢測、電子商務網站用戶點擊流流數據具有如下特征：數據快速持續到達，潛在大小也許是無窮無盡的數據來源眾多，格式復雜

數據量大，但是不十分關注存儲，一旦經過處理，要么被丟棄，要么被歸檔存儲注重數據的整體價值，不過分關注個別數據

數據順序顛倒，或者不完整，系統無法控制將要處理的新到達的數據元素的順序8.1.1靜態數據和流數據《大數據》xxu@華東師范大學信第息六管頁理，共系七十三頁。

許鑫

對靜態數據和流數據的處理，對應著兩種截然不同的計算模式：批量計算和實時計算8.1.2批量計算和實時計算圖8-2

數據的兩種處理模型批量計算：充裕時間處理靜態數據，如

Hadoop流數據不適合采用批量計算，因為流數據不適合用傳統的關系模型建模流數據必須采用實時計算，響應時間為秒級數據量少時，不是問題，但是，在大數據時代，數據格式復雜、來源眾多、數據量巨大，對實時計算提出了很大的挑戰。因此，針對流數據的實時計算——流計算，應運而生《大數據》xxu@華東師范大學信第息七管頁理，共系七十三頁。

許鑫

流計算：實時獲取來自不同數據源的海量數據，經過實時分析處理，獲得有價值的信息8.1.3流計算概念圖8-3流計算示意圖《大數據》xxu@華東師范大學信第息八管頁理，共系七十三頁。

許鑫

流計算秉承一個基本理念，即數據的價值隨著時間的流逝而降低，如用戶點擊流。因此，當事件出現時就應該立即進行處理，而不是緩存起來進行批量處理。為了及時處理流數據，就需要一個低延遲、可擴展、高可靠的處理引擎對于一個流計算系統來說，它應達到如下需求：高性能：處理大數據的基本要求，如每秒處理幾十萬條數據海量式：支持TB級甚至是PB級的數據規模實時性：保證較低的延遲時間，達到秒級別，甚至是毫秒級別分布式：支持大數據的基本架構，必須能夠平滑擴展易用性：能夠快速進行開發和部署可靠性：能可靠地處理流數據8.1.3流計算概念《大數據》xxu@華東師范大學信第息九管頁理，共系七十三頁。

許鑫

Hadoop設計的初衷是面向大規模數據的批量處理，每臺機器并行運行MapReduce任務，最后對結果進行匯總輸出

MapReduce是專門面向靜態數據的批量處理的，內部各種實現機制都為批處理做了高度優化，不適合用于處理持續到達的動態數據

可能會想到一種“變通”的方案來降低批處理的時間延遲——將基于

MapReduce的批量處理轉為小批量處理，將輸入數據切成小的片段，每隔一個周期就啟動一次MapReduce作業。但這種方式也無法有效處理流數據

切分成小片段，可以降低延遲，但是也增加了附加開銷，還要處理片段之間依賴關系需要改造MapReduce以支持流式處理8.1.4流計算與Hadoop結論：魚和熊掌不可兼得，Hadoop擅長批處理，不適合流計算《大數據》xxu@華東師范大學信第息十管頁理，共系七十三頁。

許鑫當前業界誕生了許多專門的流數據實時計算系統來滿足各自需求

目前有三類常見的流計算框架和平臺：商業級的流計算平臺、開源流計算框架、公司為支持自身業務開發的流計算框架商業級：IBM

InfoSphere

Streams和IBM

StreamBase較為常見的是開源流計算框架，代表如下：

TwitterStorm：免費、開源的分布式實時計算系統，可簡單、高效、可靠地處理大量的流數據

Yahoo!S4（Simple

Scalable

Streaming

System）：開源流計算平臺，是通用的、分布式的、可擴展的、分區容錯的、可插拔的流式系統公司為支持自身業務開發的流計算框架：Facebook

PumaDstream（百度）銀河流數據處理平臺（淘寶）8.1.5流計算框架《大數據》許鑫xxu@華東師范大學信第息十管一頁理，系共七十三頁。8.2

流計算處理流程?8.2.1概述?8.2.2數據實時采集?8.2.3數據實時計算?8.2.4實時查詢服務《大數據》許鑫xxu@華東師范大學信第息十管二頁理，系共七十三頁。傳統的數據處理流程，需要先采集數據并存儲在關系數據庫等數據管理系統中，之后由用戶通過查詢操作和數據管理系統進行交互傳統的數據處理流程示意圖傳統的數據處理流程隱含了兩個前提：

存儲的數據是舊的。存儲的靜態數據是過去某一時刻的快照，這些數據在查詢時可能已不具備時效性了需要用戶主動發出查詢來獲取結果8.2.1數據處理流程《大數據》許鑫xxu@華東師范大學信第息十管三頁理，系共七十三頁。流計算的處理流程一般包含三個階段：數據實時采集、數據實時計算、實時查詢服務8.2.1數據處理流程流計算處理流程示意圖《大數據》許鑫xxu@華東師范大學信第息十管四頁理，系共七十三頁。

數據實時采集階段通常采集多個數據源的海量數據，需要保證實時性、低延遲與穩定可靠

以日志數據為例，由于分布式集群的廣泛應用，數據分散存儲在不同的機器上，因此需要實時匯總來自不同機器上的日志數據

目前有許多互聯網公司發布的開源分布式日志采集系統均可滿足每秒數百

MB的數據采集和傳輸需求，如：Facebook的ScribeLinkedIn的Kafka淘寶的Time

Tunnel基于Hadoop的Chukwa和Flume8.2.2數據實時采集《大數據》許鑫xxu@華東師范大學信第息十管五頁理，系共七十三頁。數據采集系統的基本架構一般有以下三個部分：Agent：主動采集數據，并把數據推送到Collector部分Collector：接收多個Agent的數據，并實現有序、可靠、高性能的轉發Store：存儲Collector轉發過來的數據（對于流計算不存儲數據）8.2.2數據實時采集數據采集系統基本架構《大數據》許鑫xxu@華東師范大學信第息十管六頁理，系共七十三頁。數據實時計算階段對采集的數據進行實時的分析和計算，并反饋實時結果

經流處理系統處理后的數據，可視情況進行存儲，以便之后再進行分析計算。在時效性要求較高的場景中，處理之后的數據也可以直接丟棄8.2.3數據實時計算數據實時計算流程《大數據》許鑫xxu@華東師范大學信第息十管七頁理，系共七十三頁。實時查詢服務：經由流計算框架得出的結果可供用戶進行實時查詢、展示或儲存

傳統的數據處理流程，用戶需要主動發出查詢才能獲得想要的結果。而在流處理流程中，實時查詢服務可以不斷更新結果，并將用戶所需的結果實時推送給用戶

雖然通過對傳統的數據處理系統進行定時查詢，也可以實現不斷地更新結果和結果推送，但通過這樣的方式獲取的結果，仍然是根據過去某一時刻的數據得到的結果，與實時結果有著本質的區別8.2.3實時查詢服務《大數據》許鑫xxu@華東師范大學信第息十管八頁理，系共七十三頁。可見，流處理系統與傳統的數據處理系統有如下不同：

流處理系統處理的是實時的數據，而傳統的數據處理系統處理的是預先存儲好的靜態數據

用戶通過流處理系統獲取的是實時結果，而通過傳統的數據處理系統，獲取的是過去某一時刻的結果

流處理系統無需用戶主動發出查詢，實時查詢服務可以主動將實時結果推送給用戶8.2.3實時查詢服務《大數據》許鑫xxu@華東師范大學信第息十管九頁理，系共七十三頁。

流計算是針對流數據的實時計算，可以應用在多種場景中，如Web服務、機器翻譯、廣告投放、自然語言處理、氣候模擬預測等

如百度、淘寶等大型網站中，每天都會產生大量流數據，包括用戶的搜索內容、用戶的瀏覽記錄等數據。采用流計算進行實時數據分析，可以了解每個時刻的流量變化情況，甚至可以分析用戶的實時瀏覽軌跡，從而進行實時個性化內容推薦

但是，并不是每個應用場景都需要用到流計算的。流計算適合于需要處理持續到達的流數據、對數據處理有較高實時性要求的場景8.3流計算的應用《大數據》許鑫xxu@華東師范大學信第息二管十頁理，系共七十三頁。

傳統的業務分析一般采用分布式離線計算的方式，即將數據全部保存起來，然后每隔一定的時間進行離線分析來得到結果。但這樣會導致一定的延時，難以保證結果的實時性

隨著分析業務對實時性要求的提升，離線分析模式已經不適合用于流數據的分析，也不適用于要求實時響應的互聯網應用場景

如淘寶網“雙十一”、“雙十二”的促銷活動，商家需要根據廣告效果來即時調整廣告，這就需要對廣告的受訪情況進行分析。但以往采用分布式離線分析，需要幾小時甚至一天的延時才能得到分析結果。而促銷活動只持續一天，因此，隔天才能得到的分析結果便失去了價值

雖然分布式離線分析帶來的小時級的分析延時可以滿足大部分商家的需求，但隨著實時性要求越來越高，如何實現秒級別的實時分析響應成為業務分析的一大挑戰8.3.1應用場景1:實時分析《大數據》xxu@華東師范大學信第二息十管一理頁，系共七十三頁。許鑫

針對流數據，“量子恒道”開發了海量數據實時流計算框架Super

Mario。通過該框架，量子恒道可處理每天TB級的實時流數據，并且從用戶發出請求到數據展示，整個延時控制在2-3秒內，達到了實時性的要求8.3.1應用場景1:實時分析Super

Mario處理流程《大數據》xxu@華東師范大學信第二息十管二理頁，系共七十三頁。許鑫流計算不僅為互聯網帶來改變，也能改變我們的生活

如提供導航路線，一般的導航路線并沒有考慮實時的交通狀況，即便在計算路線時有考慮交通狀況，往往也只是使用了以往的交通狀況數據。要達到根據實時交通狀態進行導航的效果，就需要獲取海量的實時交通數據并進行實時分析

借助于流計算的實時特性，不僅可以根據交通情況制定路線，而且在行駛過程中，也可以根據交通情況的變化實時更新路線，始終為用戶提供最佳的行駛路線8.3.1應用場景2:實時交通《大數據》xxu@華東師范大學信第二息十管三理頁，系共七十三頁。許鑫?8.4.1Storm簡介?8.4.2Storm的特點?8.4.3Storm設計思想?8.4.4Storm框架設計8.4開源流計算框架Storm《大數據》xxu@華東師范大學信第二息十管四理頁，系共七十三頁。許鑫

以前只有政府機構和金融機構能夠通過昂貴的定制系統來滿足流數據實時分析計算需求

早期對于流計算的研究多數是基于對傳統數據庫處理的流式化，即實時數據庫，很少研究流計算框架Yahoo!

S4和Twitter

Storm的開源，改變了這個情況在流數據處理上比MapReduce更有優勢批處理系統關注吞吐率，流處理系統關注延時Yahoo!S4和Twitter

Storm改變了開發實時應用的方式以前既要關注處理邏輯，還要解決實時數據獲取、傳輸、存儲現在可以快速低成本搭建起實時流處理系統8.4開源流計算框架Storm《大數據》xxu@華東師范大學信第二息十管五理頁，系共七十三頁。許鑫

Twitter

Storm是一個免費、開源的分布式實時計算系統，Storm對于實時計算的意義類似于Hadoop對于批處理的意義，Storm可以簡單、高效、可靠地處理流

數據，并支持多種編程語言

Storm框架可以方便地與數據庫系統進行整合，從而開發出強大的實時計算系統8.4.1

Storm簡介《大數據》xxu@華東師范大學信第二息十管六理頁，系共七十三頁。許鑫

Twitter是全球訪問量最大的社交網站之一，Twitter開發Storm流處理框架也是為了應對其不斷增長的流數據實時處理需求8.4.1

Storm簡介Twitter的分層數據處理架構《大數據》xxu@華東師范大學信第二息十管七理頁，系共七十三頁。許鑫8.4.2

Storm的特點

Storm可用于許多領域中，如實時分析、在線機器學習、持續計算、遠程RPC、數據提取加載轉換等Storm具有以下主要特點：整合性：Storm可方便地與隊列系統和數據庫系統進行整合簡易的API：Storm的API在使用上即簡單又方便可擴展性：Storm的并行特性使其可以運行在分布式集群中容錯性：Storm可自動進行故障節點的重啟、任務的重新分配可靠的消息處理：Storm保證每個消息都能完整處理支持各種編程語言：Storm支持使用各種編程語言來定義任務快速部署：Storm可以快速進行部署和使用免費、開源：Storm是一款開源框架，可以免費使用《大數據》xxu@華東師范大學信第二息十管八理頁，系共七十三頁。許鑫8.4.3

Storm設計思想Storm主要術語包括Streams、Spouts、Bolts、Topology和Stream

GroupingsStreams：Storm將流數據Stream描述成一個無限的Tuple序列，這些Tuple序列會以分布式的方式并行地創建和處理每個tuple是一堆值，每個值有一個名字，并且每個值可以是任何類型Tuple本來應該是一個Key-Value的Map，由于各個組件間傳遞的tuple的字段名稱已經事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個Value

List（值列表）Field1Field2Field3Field4《大數據》xxu@華東師范大學信第二息十管九理頁，系共七十三頁。許鑫8.4.3

Storm設計思想Spout：Storm認為每個Stream都有一個源頭，并把這個源頭抽象為Spout

通常Spout會從外部數據源（隊列、數據庫等）讀取數據，然后封裝成

Tuple形式，發送到Stream中。Spout是一個主動的角色，在接口內部有個nextTuple函數，Storm框架會不停的調用該函數《大數據》許鑫xxu@華東師范大學信第息三管十頁理，系共七十三頁。8.4.3

Storm設計思想Bolt：Storm將Streams的狀態轉換過程抽象為Bolt。Bolt即可以處理Tuple，也可以將處理后的Tuple作為新的Streams發送給其他BoltBolt可以執行過濾、函數操作、Join、操作數據庫等任何操作

Bolt是一個被動的角色，其接口中有一個execute(Tuple

input)方法，在接收到消息之后會調用此函數，用戶可以在此方法中執行自己的處理邏輯《大數據》8.4.3

Storm設計思想Topology：Storm將Spouts和Bolts組成的網絡抽象成Topology，它可以被提交到Storm集群執行。Topology可視為流轉換圖，圖中節點是一個Spout或Bolt，邊則表示Bolt訂閱了哪個Stream。當Spout或者Bolt發送元組時，它會把元組發送到每個訂閱了該Stream的Bolt上進行處理

Topology里面的每個處理組件（Spout或Bolt）都包含處理邏輯，而組件之間的連接則表示數據流動的方向Topology里面的每一個組件都是并行運行的在Topology里面可以指定每個組件的并行度，Storm會在集群里面分配那么多的線程來同時計算在Topology的具體實現上，Storm中的

Topology定義僅僅是一些Thrift結構體（二進制高性能的通信中間件），支持各種編程語言進行定義SpoutSpoutBoltBoltBoltxxu@BoltBolt華東師范大學信第三息十管一理頁，系共七十三頁。許鑫《大數據》xxu@華東師范大學信第三息十管二理頁，系共七十三頁。許鑫8.4.3

Storm設計思想Stream

Groupings：S

torm中的Stream

Groupings用于告知Topology如何在

兩個組件間（如Spout和Bolt之間，或者不同的Bolt之間）進行Tuple的傳送。每一個Spout和Bolt都可以有多個分布式任務，一個任務在什么時候、以什么方式發送Tuple就是由Stream

Groupings來決定的《大數據》xxu@華東師范大學信第三息十管三理頁，系共七十三頁。許鑫8.4.3

Storm設計思想目前，Storm中的Stream

Groupings有如下幾種方式：(1)ShuffleGrouping：隨機分組，隨機分發Stream中的Tuple，保證每個Bolt的Task接收Tuple數量大致一致(2)FieldsGrouping：按照字段分組，保證相同字段的Tuple分配到同一個Task中(3)AllGrouping：廣播發送，每一個Task都會收到所有的Tuple(4)GlobalGrouping：全局分組，所有的Tuple都發送到同一個Task中(5)NonGrouping：不分組，和ShuffleGrouping類似，當前Task的執行會和它的被訂閱者在同一個線程中執行(6)DirectGrouping：直接分組，直接指定由某個Task來執行Tuple的處理《大數據》xxu@華東師范大學信第三息十管四理頁，系共七十三頁。許鑫8.4.4

Storm框架設計HadoopStorm應用名稱JobTopology系統角色JobTrackerNimbusTaskTrackerSupervisor組件接口Map/ReduceSpout/BoltStorm和Hadoop架構組件功能對應關系Storm運行任務的方式與Hadoop類似：Hadoop運行的是MapReduce作業，而Storm運行的是“Topology”但兩者的任務大不相同，主要的不同是：MapReduce作業最終會完成計算并結束運行，而Topology將持續處理消息（直到人為終止）《大數據》xxu@華東師范大學信第三息十管五理頁，系共七十三頁。許鑫8.4.4

Storm框架設計Storm集群采用“Master—Worker”的節點方式：

Master節點運行名為“Nimbus”的后臺程序（類似Hadoop中的

“JobTracker”），負責在集群范圍內分發代碼、為Worker分配任務和監測故障

Worker節點運行名為“Supervisor”的后臺程序，負責監聽分配給它所在機器的工作，即根據Nimbus分配的任務來決定啟動或停止Worker進程，一個

Worker節點上同時運行若干個Worker進程《大數據》xxu@華東師范大學信第三息十管六理頁，系共七十三頁。許鑫

Storm使用Zookeeper來作為分布式協調組件，負責Nimbus和多個Supervisor之間的所有協調工作。借助于Zookeeper，若Nimbus進程或Supervisor進程意外終止，重啟時也能讀取、恢復之前的狀態并繼續工作，使得Storm極其穩定8.4.4

Storm框架設計Storm集群架構示意圖Worker進程Worker進程……Worker進程《大數據》xxu@華東師范大學信第三息十管七理頁，系共七十三頁。許鑫8.4.4

Storm框架設計worker:每個worker進程都屬于一個特定的Topology，每個Supervisor節點的worker可以有多個，每個worker對Topology中的每個組件（Spout或Bolt）運行一個或者多個

executor線程來提供task的運行服務executor：executor是產生于worker進程內部的線程，會執行同一個組件的一個或者多個task。task:實際的數據處理由task完成，在Topology的生命周期中，每個組件的task數目是不會發生變化的，而executor的數目卻不一定。executor數目小于等于task的數目，默認情況下，二者是相等的Worker、Executor和Task的關系《大數據》xxu@華東師范大學信第三息十管八理頁，系共七十三頁。許鑫8.4.4

Storm框架設計Storm工作流程示意圖基于這樣的架構設計，Storm的工作流程如下圖所示：所有Topology任務的提交必須在Storm客戶端節點上進行，提交后，由Nimbus節點分配給其他Supervisor節點進行處理Nimbus節點首先將提交的Topology進行分片，分成一個個Task，分配給相應的Supervisor，并將Task和Supervisor相關的信息提交到Zookeeper集群上Supervisor會去Zookeeper集群上認領自己的Task，通知自己的Worker進程進行Task的處理說明：在提交了一個Topology之后，Storm就會創建Spout/Bolt實例并進行序列化。之后，將序列化的組件發送給所有的任務所在的機器(即Supervisor節點)，在每一個任務上反序列化組件《大數據》xxu@華東師范大學信第三息十管九理頁，系共七十三頁。許鑫8.5

Spark

StreamingSpark

Streaming設計Spark

Streaming與Storm的對比《大數據》許鑫xxu@華東師范大學信第息四管十頁理，系共七十三頁。8.5.1

Spark

Streaming設計SparkStreaming可整合多種輸入數據源，如Kafka、Flume、HDFS，甚至是普通的TCP套接字。經處理后的數據可存儲至文件系統、數據庫，或顯示在儀表盤里圖13

Spark

Streaming支持的輸入、輸出數據源《大數據》xxu@華東師范大學信第四息十管一理頁，系共七十三頁。許鑫8.5.1

Spark

Streaming設計Spark

Streaming的基本原理是將實時輸入數據流以時間片（秒級）為單位進行拆分，然后經

Spark引擎以類似批處理的方式處理每個時間片數據圖14

Spark

Streaming執行流程《大數據》xxu@華東師范大學信第四息十管二理頁，系共七十三頁。許鑫8.5.1

Spark

Streaming設計Spark

Streaming最主要的抽象是DStream（Discretized

Stream，離散化數據流），表示連續不斷的數據流。在內部實現上，Spark

Streaming的輸入數據按照時間片（如1秒）分成一段一

段的DStream，每一段數據轉換為Spark中的RDD，并且對DStream的操作都最終轉變為對相應的RDD的操作圖15

DStream操作示意圖《大數據》xxu@華東師范大學信第四息十管三理頁，系共七十三頁。許鑫8.5.2

Spark

Streaming與Storm的對比SparkStreaming和Storm最大的區別在于，SparkStreaming無法實現毫秒級的流計算，而Storm可以實現毫秒級響應Spark

Streaming構建在Spark上，一方面是因為Spark的低延遲執行引擎（100ms+）可以用于實時計算，另一方面，相比于Storm，RDD數據集更容易做高效的容錯處理Spark

Streaming采用的小批量處理的方式使得它可以同時兼容批量和實時數據處理的邏輯和算法，因此，方便了一些需要歷史數據和實時數據聯合分析的特定應用場合《大數據》xxu@華東師范大學信第四息十管四理頁，系共七十三頁。許鑫8.6

Samza基本概念系統架構《大數據》xxu@華東師范大學信第四息十管五理頁，系共七十三頁。許鑫8.6.1基本概念1.作業一個作業（Job）是對一組輸入流進行處理轉化成輸出流的程序。《大數據》xxu@華東師范大學信第四息十管六理頁，系共七十三頁。許鑫8.6.1基本概念分區Samza的流數據單位既不是Storm中的元組，也不是Spark

Streaming中的

DStream，而是一條條消息Samza中的每個流都被分割成一個或多個分區，對于流里的每一個分區而言，都是一個有序的消息序列，后續到達的消息會根據一定規則被追加到其中一個分區里《大數據》xxu@華東師范大學信第四息十管七理頁，系共七十三頁。許鑫8.6.1基本概念任務一個作業會被進一步分割成多個任務（Task）來執行，其中，每個任務負責處理作業中的一個分區分區之間沒有定義順序，從而允許每一個任務獨立執行YARN調度器負責把任務分發給各個機器，最終，一個工作中的多個任務會被分發到多個機器進行分布式并行處理《大數據》xxu@華東師范大學信第四息十管八理頁，系共七十三頁。許鑫8.6.1基本概念數據流圖一個數據流圖是由多個作業構成的，其中，圖中的每個節點表示包含數據的流，每條邊表示數據傳輸多個作業串聯起來就完成了流式的數據處理流程由于采用了異步的消息訂閱分發機制，不同任務之間可以獨立運行《大數據》xxu@華東師范大學信第四息十管九理頁，系共七十三頁。許鑫8.6.2系統架構MapReduce批處理架構Samza流處理架構數據層HDFSKafka執行層YARNYARN處理層MapReduceSamza

API表MapReduce批處理架構和Samza流處理架構的類比Samza系統架構主要包括–流數據層（Kafka）–執行層（YARN）–處理層（Samza

API）流處理層和執行層都被設計成可插拔的，開發人員可以使用其他框架來替代YARN和Kafka《大數據》許鑫xxu@華東師范大學信第息五管十頁理，系共七十三頁。8.6.2系統架構處理分析過程如下：Samza客戶端需要執行一個Samza作業時，它會向YARN的ResouceManager提交作業請求ResouceManager通過與NodeManager溝通為該作業分配容器（包含了CPU、內存等資源）來運行SamzaApplicationMasterSamza

ApplicationMaster進一步向ResourceManager申請運行任務的容器獲得容器后，Samza

ApplicationMaster與容器所在的

NodeManager溝通，啟動該容器，并在其中運

行Samza

Task

RunnerSamza

Task

Runner負責執行具體的Samza任務，完成流數據處理分析《大數據》xxu@華東師范大學信第五息十管一理頁，系共七十三頁。許鑫8.7

Storm、Spark

Streaming和Samza的應用場景從編程的靈活性來講，Storm是比較理想的選擇，它使用Apache

Thrift，可以用任何編程語言來編寫拓撲結構（Topology）當需要在一個集群中把流計算和圖計算、機器學習、SQL查詢分析等進行結合時，可以選擇Spark

Streaming，因為，在Spark上可以統一部署Spark

SQL，Spark

Streaming、MLlib，GraphX等組件，提供便捷的一體化編程模型當有大量的狀態需要處理時，比如每個分區都有數十億個元組，則可以選擇Samza。當應用場景需要毫秒級響應時，可以選擇Storm和Samza，因為Spark

Streaming無法實現毫秒級的流計算《大數據》xxu@華東師范大學信第五息十管二理頁，系共七十三頁。許鑫8.8

Storm編程實踐編寫Storm程序安裝Storm的基本過程運行Storm程序Storm上機實踐詳細過程，請參考廈門大學數據庫實驗室建設的

“中國高校大數據課程公共服務平臺”中的“大數據課程學生服務站”中的“學習指南”欄目：學生服務站地址：/post/4331/學習指南欄目中包含了《第八章流計算學習指南》/blog/767/掃一掃訪問學生服務站《大數據》xxu@華東師范大學信第五息十管三理頁，系共七十三頁。許鑫8.8.1編寫Storm程序程序任務：單詞統計

基于Storm的單詞統計在形式上與基于MapReduce的單詞統計是類似的，

MapReduce使用的是Map和Reduce的抽象，而Storm使用的是Spout和Bolt的抽象Storm進行單詞統計的整個流程：從Spout中發送Stream（每個英文句子為一個Tuple）用于分割單詞的Bolt將接收的句子分解為獨立的單詞，將單詞作為Tuple的字段名發送出去用于計數的Bolt接收表示單詞的Tuple，并對其進行統計輸出每個單詞以及單詞出現過的次數《大數據》xxu@華東師范大學信第五息十管四理頁，系共七十三頁。許鑫8.8.1編寫Storm程序一個句子經Storm的單詞統計得出的結果《大數據》xxu@華東師范大學信第五息十管五理頁，系共七十三頁。許鑫Storm的編程模型非常簡單，如下代碼即定義了整個單詞統計Topology的整體邏輯import

org.apache.storm.Config;Import

……public

class

WordCountTopology

{public

static

class

RandomSentenceSpout

extends

BaseRichSpout

{……}public

static

class

SplitSentence

extends

ShellBolt

implements

IRichBolt

{……}public

static

class

WordCount

extends

BaseBasicBolt

{……}public

static

void

main(String[]

args)

throws

Exception

{TopologyBuilder

builder

=

new

TopologyBuilder();builder.setSpout("sentences",

new

RandomSentenceSpout(),

5);builder.setBolt("split",

new

SplitSentence(),

8).shuffleGrouping("sentences");builder.setBolt("count",

new

WordCount(),

12).fieldsGrouping("split",

new

Fields("word"));……}}8.8.1編寫Storm程序《大數據》xxu@華東師范大學信第五息十管六理頁，系共七十三頁。許鑫8.8.1編寫Storm程序main()函數中的處理邏輯Topology中僅定義了整體的計算邏輯，還需要定義具體的處理函數。具體的處理函數可以使用任一編程語言來定義，甚至也可以結合多種編程語言來實現《大數據》xxu@華東師范大學信第五息十管七理頁，系共七十三頁。許鑫RandomSentenceSpout類備注：為簡單起見，RandomSentenceSpout省略了類中的一些方法public

class

RandomSentenceSpout

extends

BaseRichSpout

{SpoutOutputCollector_collector;Random

_rand;@Overridepublic

void

nextTuple()

{Utils.sleep(100);String[]

sentences

=

new

String[]{

"the

cow

jumped

over

the

moon",

"an

apple

a

day

keeps

the

doctor

away","four

score

and

seven

years

ago",

"snow

white

and

the

seven

dwarfs",

"i

am

at

two

with

nature"

};String

sentence

=

sentences[_rand.nextInt(sentences.length)];_collector.emit(new

Values(sentence));}@Overridepublic

void

declareOutputFields(OutputFieldsDeclarer

declarer)

{declarer.declare(new

Fields(“sentences"));}}8.8.1編寫Storm程序《大數據》xxu@華東師范大學信第五息十管八理頁，系共七十三頁。許鑫8.8.1編寫Storm程序SplitSentence類

如SplitSentence()方法雖然是通過Java語言定義的，但具體的操作可通過Python腳本來完成Topology里面的每個組件必須定義它要發射的Tuple的每個字段《大數據》xxu@華東師范大學信第五息十管九理頁，系共七十三頁。許鑫8.8.1編寫Storm程序splitsentence.py

Python腳本splitsentence.py定義了一個簡單的單詞分割方法，即通過空格來分割單詞。分割后的單詞通過emit()方法以Tuple的形式發送給訂閱了該Stream的Bolt進行接收和處理《大數據》許鑫xxu@華東師范大學信第息六管十頁理，系共七十三頁。8.8.1編寫Storm程序WordCount類單詞統計的具體邏輯：首先判斷單詞是否統計過，若未統計過，需先將count值置為0。若單詞已統計過，則每出現一次該單詞，count值就加1《大數據》xxu@華東師范大學信第六息十管一理頁，系共七十三頁。許鑫8.8.1編寫Storm程序

上述雖然是一個簡單的單詞統計，但對其進行擴展，便可應用到許多場景中，如微博中的實時熱門話題。Twitter也正是使用了Storm框架實現了實時熱門話題Twitter實時熱門話題處理流程示意圖《大數據》xxu@華東師范大學信第六息十管二理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程安裝Storm的基本過程如下：第一步：安裝Java環境第二步：安裝

Zookeeper第三步：安裝Storm（單機）第四步：關閉Storm本實例中Storm具體運行環境如下：CentOS

6.4Storm

0.9.6Java

JDK

1.7ZooKeeper

3.4.6Python

2.6備注：CentOS中已默認安裝了Python

2.6，我們還需要安裝JDK

環境以及分布式應用程序協調服務ZookeeperStorm上機實踐詳細過程，請參考廈門大學數據庫實驗室建設的“中國高校大數據課程公共服務平臺”中的“大數據課程學生服務站”中的“學習指南”欄目：學生服務站地址：/post/4331/學習指南欄目中包含了《第八章流計算學習指南》/blog/767/《大數據》xxu@華東師范大學信第六息十管三理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第一步：安裝Java環境Storm

運行需要Java環境，可選擇Oracle

的JDK，或是OpenJDK，現在一般

Linux

系統默認安裝的基本是

OpenJDK，如CentOS

6.4

就默認安裝了OpenJDK1.7。但需要注意的是，CentOS

6.4

中默認安裝的只是Java

JRE，而不是

JDK，為了開發方便，我們還是需要通過yum

進行安裝JDK接著需要配置一下JAVA_HOME

環境變量，為方便，可以在~/.bashrc

中進行設置《大數據》xxu@華東師范大學信第六息十管四理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第二步：安裝Zookeeper到官網下載Zookeeper，比如下載“zookeeper-3.4.6.tar.gz”下載后執行如下命令進行安裝zookeeper（將命令中3.4.6改為你下載的版本）：chown命令讓hadoop用戶擁有zookeeper目錄下的所有文件的權限《大數據》xxu@華東師范大學信第六息十管五理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第二步：安裝Zookeeper（續）接著執行如下命令進行zookeeper配置：進入zoo.cfg文件編輯狀態后，將當中的dataDir=/tmp/zookeeper更改為dataDir=/usr/local/zookeeper/tmp

。接著執行：《大數據》xxu@華東師范大學信第六息十管六理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第三步：安裝Storm（單機）到官網下載Storm，比如Storm0.9.6下載后執行如下命令進行安裝Storm：接著執行如下命令進行Storm配置:備注：storm的運行有兩種模式:本地模式和分布式模式.在本地模式中，storm用一個進程里面的線程來模擬所有的spout和bolt.本地模式對開發和測試來說比較有用。在分布式模式下，storm由一堆機器組成。當提交topology給master的時候，master負責分發代碼并且負責給topolgoy分配工作進程。如果一個工作進程掛掉了，master節點會把它重新分配到其它節點《大數據》xxu@華東師范大學信第六息十管七理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第三步：安裝Storm（單機）

（續）修改其中的storm.zookeeper.servers和nimbus.host

兩個配置項，即取消掉注釋且都修改值為（我們只需要在單機上運行），如下圖所示。然后就可以啟動Storm

了。執行如下命令啟動nimbus

后臺進程：《大數據》xxu@華東師范大學信第六息十管八理頁，系共七十三頁。許鑫8.8.2安裝Storm的基本過程第三步：安裝Storm（單機）

（續）啟動nimbus

后，終端被該進程占用了，不能再繼續執行其他命令了。因此我們需要另外開啟一個終