商業(yè)大數(shù)據(jù)分析課件

2022/12/17商業(yè)大數(shù)據(jù)分析2022/12/17商業(yè)大數(shù)據(jù)分析1提綱8.1流計算概述8.2流計算處理流程8.3流計算應(yīng)用8.4流計算開源框架–Storm8.5SparkStreaming8.6Samza8.7Storm、SparkStreaming和Samza的應(yīng)用場景8.8Storm編程實踐提綱8.1流計算概述28.1 流計算概述8.1.1 靜態(tài)數(shù)據(jù)和流數(shù)據(jù)8.1.2 批量計算和實時計算8.1.3 流計算概念8.1.4 流計算與Hadoop8.1.5 流計算框架8.1 流計算概述8.1.1 靜態(tài)數(shù)據(jù)和流數(shù)據(jù)3很多企業(yè)為了支持決策分析而構(gòu)建的數(shù)據(jù)倉庫系統(tǒng)，其中存放的大量歷史數(shù)據(jù)就是靜態(tài)數(shù)據(jù)。技術(shù)人員可以利用數(shù)據(jù)挖掘和OLAP（On-LineAnalyticalProcessing）分析工具從靜態(tài)數(shù)據(jù)中找到對企業(yè)有價值的信息8.1.1靜態(tài)數(shù)據(jù)和流數(shù)據(jù)很多企業(yè)為了支持決策分析而構(gòu)建的數(shù)據(jù)倉庫系統(tǒng)，其中存放的大量4近年來，在Web應(yīng)用、網(wǎng)絡(luò)監(jiān)控、傳感監(jiān)測等領(lǐng)域，興起了一種新的數(shù)據(jù)密集型應(yīng)用——流數(shù)據(jù)，即數(shù)據(jù)以大量、快速、時變的流形式持續(xù)到達實例：PM2.5檢測、電子商務(wù)網(wǎng)站用戶點擊流流數(shù)據(jù)具有如下特征：數(shù)據(jù)快速持續(xù)到達，潛在大小也許是無窮無盡的數(shù)據(jù)來源眾多，格式復(fù)雜數(shù)據(jù)量大，但是不十分關(guān)注存儲，一旦經(jīng)過處理，要么被丟棄，要么被歸檔存儲注重數(shù)據(jù)的整體價值，不過分關(guān)注個別數(shù)據(jù)數(shù)據(jù)順序顛倒，或者不完整，系統(tǒng)無法控制將要處理的新到達的數(shù)據(jù)元素的順序8.1.1靜態(tài)數(shù)據(jù)和流數(shù)據(jù)近年來，在Web應(yīng)用、網(wǎng)絡(luò)監(jiān)控、傳感監(jiān)測等領(lǐng)域，興起了一種新5對靜態(tài)數(shù)據(jù)和流數(shù)據(jù)的處理，對應(yīng)著兩種截然不同的計算模式：批量計算和實時計算8.1.2批量計算和實時計算圖8-2數(shù)據(jù)的兩種處理模型批量計算：充裕時間處理靜態(tài)數(shù)據(jù)，如Hadoop流數(shù)據(jù)不適合采用批量計算，因為流數(shù)據(jù)不適合用傳統(tǒng)的關(guān)系模型建模流數(shù)據(jù)必須采用實時計算，響應(yīng)時間為秒級數(shù)據(jù)量少時，不是問題，但是，在大數(shù)據(jù)時代，數(shù)據(jù)格式復(fù)雜、來源眾多、數(shù)據(jù)量巨大，對實時計算提出了很大的挑戰(zhàn)。因此，針對流數(shù)據(jù)的實時計算——流計算，應(yīng)運而生對靜態(tài)數(shù)據(jù)和流數(shù)據(jù)的處理，對應(yīng)著兩種截然不同的計算模式：批量6流計算：實時獲取來自不同數(shù)據(jù)源的海量數(shù)據(jù)，經(jīng)過實時分析處理，獲得有價值的信息8.1.3流計算概念圖8-3流計算示意圖流計算：實時獲取來自不同數(shù)據(jù)源的海量數(shù)據(jù)，經(jīng)過實時分析處理，7流計算秉承一個基本理念，即數(shù)據(jù)的價值隨著時間的流逝而降低，如用戶點擊流。因此，當(dāng)事件出現(xiàn)時就應(yīng)該立即進行處理，而不是緩存起來進行批量處理。為了及時處理流數(shù)據(jù)，就需要一個低延遲、可擴展、高可靠的處理引擎對于一個流計算系統(tǒng)來說，它應(yīng)達到如下需求：高性能：處理大數(shù)據(jù)的基本要求，如每秒處理幾十萬條數(shù)據(jù)海量式：支持TB級甚至是PB級的數(shù)據(jù)規(guī)模實時性：保證較低的延遲時間，達到秒級別，甚至是毫秒級別分布式：支持大數(shù)據(jù)的基本架構(gòu)，必須能夠平滑擴展易用性：能夠快速進行開發(fā)和部署可靠性：能可靠地處理流數(shù)據(jù)8.1.3流計算概念流計算秉承一個基本理念，即數(shù)據(jù)的價值隨著時間的流逝而降低，如8Hadoop設(shè)計的初衷是面向大規(guī)模數(shù)據(jù)的批量處理，每臺機器并行運行MapReduce任務(wù)，最后對結(jié)果進行匯總輸出MapReduce是專門面向靜態(tài)數(shù)據(jù)的批量處理的，內(nèi)部各種實現(xiàn)機制都為批處理做了高度優(yōu)化，不適合用于處理持續(xù)到達的動態(tài)數(shù)據(jù)可能會想到一種“變通”的方案來降低批處理的時間延遲——將基于MapReduce的批量處理轉(zhuǎn)為小批量處理，將輸入數(shù)據(jù)切成小的片段，每隔一個周期就啟動一次MapReduce作業(yè)。但這種方式也無法有效處理流數(shù)據(jù)切分成小片段，可以降低延遲，但是也增加了附加開銷，還要處理片段之間依賴關(guān)系需要改造MapReduce以支持流式處理8.1.4流計算與Hadoop結(jié)論：魚和熊掌不可兼得，Hadoop擅長批處理，不適合流計算Hadoop設(shè)計的初衷是面向大規(guī)模數(shù)據(jù)的批量處理，每臺機器并9當(dāng)前業(yè)界誕生了許多專門的流數(shù)據(jù)實時計算系統(tǒng)來滿足各自需求目前有三類常見的流計算框架和平臺：商業(yè)級的流計算平臺、開源流計算框架、公司為支持自身業(yè)務(wù)開發(fā)的流計算框架商業(yè)級：IBMInfoSphereStreams和IBMStreamBase較為常見的是開源流計算框架，代表如下：TwitterStorm：免費、開源的分布式實時計算系統(tǒng)，可簡單、高效、可靠地處理大量的流數(shù)據(jù)Yahoo!S4（SimpleScalableStreamingSystem）：開源流計算平臺，是通用的、分布式的、可擴展的、分區(qū)容錯的、可插拔的流式系統(tǒng)公司為支持自身業(yè)務(wù)開發(fā)的流計算框架：FacebookPumaDstream（百度）銀河流數(shù)據(jù)處理平臺（淘寶）8.1.5流計算框架當(dāng)前業(yè)界誕生了許多專門的流數(shù)據(jù)實時計算系統(tǒng)來滿足各自需求8.108.2 流計算處理流程8.2.1 概述8.2.2 數(shù)據(jù)實時采集8.2.3 數(shù)據(jù)實時計算8.2.4 實時查詢服務(wù)8.2 流計算處理流程8.2.1 概述11傳統(tǒng)的數(shù)據(jù)處理流程，需要先采集數(shù)據(jù)并存儲在關(guān)系數(shù)據(jù)庫等數(shù)據(jù)管理系統(tǒng)中，之后由用戶通過查詢操作和數(shù)據(jù)管理系統(tǒng)進行交互傳統(tǒng)的數(shù)據(jù)處理流程隱含了兩個前提：存儲的數(shù)據(jù)是舊的。存儲的靜態(tài)數(shù)據(jù)是過去某一時刻的快照，這些數(shù)據(jù)在查詢時可能已不具備時效性了需要用戶主動發(fā)出查詢來獲取結(jié)果8.2.1數(shù)據(jù)處理流程傳統(tǒng)的數(shù)據(jù)處理流程示意圖傳統(tǒng)的數(shù)據(jù)處理流程，需要先采集數(shù)據(jù)并存儲在關(guān)系數(shù)據(jù)庫等數(shù)據(jù)管12流計算的處理流程一般包含三個階段：數(shù)據(jù)實時采集、數(shù)據(jù)實時計算、實時查詢服務(wù)8.2.1數(shù)據(jù)處理流程流計算處理流程示意圖流計算的處理流程一般包含三個階段：數(shù)據(jù)實時采集、數(shù)據(jù)實時計算13數(shù)據(jù)實時采集階段通常采集多個數(shù)據(jù)源的海量數(shù)據(jù)，需要保證實時性、低延遲與穩(wěn)定可靠以日志數(shù)據(jù)為例，由于分布式集群的廣泛應(yīng)用，數(shù)據(jù)分散存儲在不同的機器上，因此需要實時匯總來自不同機器上的日志數(shù)據(jù)目前有許多互聯(lián)網(wǎng)公司發(fā)布的開源分布式日志采集系統(tǒng)均可滿足每秒數(shù)百MB的數(shù)據(jù)采集和傳輸需求，如：Facebook的ScribeLinkedIn的Kafka淘寶的TimeTunnel基于Hadoop的Chukwa和Flume8.2.2數(shù)據(jù)實時采集數(shù)據(jù)實時采集階段通常采集多個數(shù)據(jù)源的海量數(shù)據(jù)，需要保證實時性14數(shù)據(jù)采集系統(tǒng)的基本架構(gòu)一般有以下三個部分：Agent：主動采集數(shù)據(jù)，并把數(shù)據(jù)推送到Collector部分Collector：接收多個Agent的數(shù)據(jù)，并實現(xiàn)有序、可靠、高性能的轉(zhuǎn)發(fā)Store：存儲Collector轉(zhuǎn)發(fā)過來的數(shù)據(jù)（對于流計算不存儲數(shù)據(jù)）8.2.2數(shù)據(jù)實時采集數(shù)據(jù)采集系統(tǒng)基本架構(gòu)數(shù)據(jù)采集系統(tǒng)的基本架構(gòu)一般有以下三個部分：8.2.2數(shù)據(jù)實15數(shù)據(jù)實時計算階段對采集的數(shù)據(jù)進行實時的分析和計算，并反饋實時結(jié)果經(jīng)流處理系統(tǒng)處理后的數(shù)據(jù)，可視情況進行存儲，以便之后再進行分析計算。在時效性要求較高的場景中，處理之后的數(shù)據(jù)也可以直接丟棄8.2.3數(shù)據(jù)實時計算數(shù)據(jù)實時計算流程數(shù)據(jù)實時計算階段對采集的數(shù)據(jù)進行實時的分析和計算，并反饋實時16實時查詢服務(wù)：經(jīng)由流計算框架得出的結(jié)果可供用戶進行實時查詢、展示或儲存?zhèn)鹘y(tǒng)的數(shù)據(jù)處理流程，用戶需要主動發(fā)出查詢才能獲得想要的結(jié)果。而在流處理流程中，實時查詢服務(wù)可以不斷更新結(jié)果，并將用戶所需的結(jié)果實時推送給用戶雖然通過對傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)進行定時查詢，也可以實現(xiàn)不斷地更新結(jié)果和結(jié)果推送，但通過這樣的方式獲取的結(jié)果，仍然是根據(jù)過去某一時刻的數(shù)據(jù)得到的結(jié)果，與實時結(jié)果有著本質(zhì)的區(qū)別8.2.3實時查詢服務(wù)實時查詢服務(wù)：經(jīng)由流計算框架得出的結(jié)果可供用戶進行實時查詢、17可見，流處理系統(tǒng)與傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)有如下不同：流處理系統(tǒng)處理的是實時的數(shù)據(jù)，而傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)處理的是預(yù)先存儲好的靜態(tài)數(shù)據(jù)用戶通過流處理系統(tǒng)獲取的是實時結(jié)果，而通過傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)，獲取的是過去某一時刻的結(jié)果流處理系統(tǒng)無需用戶主動發(fā)出查詢，實時查詢服務(wù)可以主動將實時結(jié)果推送給用戶8.2.3實時查詢服務(wù)可見，流處理系統(tǒng)與傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)有如下不同：8.2.318流計算是針對流數(shù)據(jù)的實時計算，可以應(yīng)用在多種場景中，如Web服務(wù)、機器翻譯、廣告投放、自然語言處理、氣候模擬預(yù)測等如百度、淘寶等大型網(wǎng)站中，每天都會產(chǎn)生大量流數(shù)據(jù)，包括用戶的搜索內(nèi)容、用戶的瀏覽記錄等數(shù)據(jù)。采用流計算進行實時數(shù)據(jù)分析，可以了解每個時刻的流量變化情況，甚至可以分析用戶的實時瀏覽軌跡，從而進行實時個性化內(nèi)容推薦但是，并不是每個應(yīng)用場景都需要用到流計算的。流計算適合于需要處理持續(xù)到達的流數(shù)據(jù)、對數(shù)據(jù)處理有較高實時性要求的場景8.3流計算的應(yīng)用流計算是針對流數(shù)據(jù)的實時計算，可以應(yīng)用在多種場景中，如Web19傳統(tǒng)的業(yè)務(wù)分析一般采用分布式離線計算的方式，即將數(shù)據(jù)全部保存起來，然后每隔一定的時間進行離線分析來得到結(jié)果。但這樣會導(dǎo)致一定的延時，難以保證結(jié)果的實時性隨著分析業(yè)務(wù)對實時性要求的提升，離線分析模式已經(jīng)不適合用于流數(shù)據(jù)的分析，也不適用于要求實時響應(yīng)的互聯(lián)網(wǎng)應(yīng)用場景如淘寶網(wǎng)“雙十一”、“雙十二”的促銷活動，商家需要根據(jù)廣告效果來即時調(diào)整廣告，這就需要對廣告的受訪情況進行分析。但以往采用分布式離線分析，需要幾小時甚至一天的延時才能得到分析結(jié)果。而促銷活動只持續(xù)一天，因此，隔天才能得到的分析結(jié)果便失去了價值雖然分布式離線分析帶來的小時級的分析延時可以滿足大部分商家的需求，但隨著實時性要求越來越高，如何實現(xiàn)秒級別的實時分析響應(yīng)成為業(yè)務(wù)分析的一大挑戰(zhàn)8.3.1應(yīng)用場景1:實時分析傳統(tǒng)的業(yè)務(wù)分析一般采用分布式離線計算的方式，即將數(shù)據(jù)全部保存20針對流數(shù)據(jù)，“量子恒道”開發(fā)了海量數(shù)據(jù)實時流計算框架SuperMario。通過該框架，量子恒道可處理每天TB級的實時流數(shù)據(jù)，并且從用戶發(fā)出請求到數(shù)據(jù)展示，整個延時控制在2-3秒內(nèi)，達到了實時性的要求8.3.1應(yīng)用場景1:實時分析SuperMario處理流程針對流數(shù)據(jù)，“量子恒道”開發(fā)了海量數(shù)據(jù)實時流計算框架Supe21流計算不僅為互聯(lián)網(wǎng)帶來改變，也能改變我們的生活如提供導(dǎo)航路線，一般的導(dǎo)航路線并沒有考慮實時的交通狀況，即便在計算路線時有考慮交通狀況，往往也只是使用了以往的交通狀況數(shù)據(jù)。要達到根據(jù)實時交通狀態(tài)進行導(dǎo)航的效果，就需要獲取海量的實時交通數(shù)據(jù)并進行實時分析借助于流計算的實時特性，不僅可以根據(jù)交通情況制定路線，而且在行駛過程中，也可以根據(jù)交通情況的變化實時更新路線，始終為用戶提供最佳的行駛路線8.3.1應(yīng)用場景2:實時交通流計算不僅為互聯(lián)網(wǎng)帶來改變，也能改變我們的生活8.3.1應(yīng)228.4.1 Storm簡介8.4.2 Storm的特點8.4.3 Storm設(shè)計思想8.4.4

Storm框架設(shè)計8.4開源流計算框架Storm8.4.1 Storm簡介8.4開源流計算框架Storm23以前只有政府機構(gòu)和金融機構(gòu)能夠通過昂貴的定制系統(tǒng)來滿足流數(shù)據(jù)實時分析計算需求早期對于流計算的研究多數(shù)是基于對傳統(tǒng)數(shù)據(jù)庫處理的流式化，即實時數(shù)據(jù)庫，很少研究流計算框架Yahoo!S4和TwitterStorm的開源，改變了這個情況在流數(shù)據(jù)處理上比MapReduce更有優(yōu)勢批處理系統(tǒng)關(guān)注吞吐率，流處理系統(tǒng)關(guān)注延時Yahoo!S4和TwitterStorm改變了開發(fā)實時應(yīng)用的方式以前既要關(guān)注處理邏輯，還要解決實時數(shù)據(jù)獲取、傳輸、存儲現(xiàn)在可以快速低成本搭建起實時流處理系統(tǒng)8.4開源流計算框架Storm以前只有政府機構(gòu)和金融機構(gòu)能夠通過昂貴的定制系統(tǒng)來滿足流數(shù)據(jù)24TwitterStorm是一個免費、開源的分布式實時計算系統(tǒng)，Storm對于實時計算的意義類似于Hadoop對于批處理的意義，Storm可以簡單、高效、可靠地處理流數(shù)據(jù)，并支持多種編程語言Storm框架可以方便地與數(shù)據(jù)庫系統(tǒng)進行整合，從而開發(fā)出強大的實時計算系統(tǒng)8.4.1Storm簡介TwitterStorm是一個免費、開源的分布式實時計算系25Twitter是全球訪問量最大的社交網(wǎng)站之一，Twitter開發(fā)Storm流處理框架也是為了應(yīng)對其不斷增長的流數(shù)據(jù)實時處理需求8.4.1Storm簡介Twitter的分層數(shù)據(jù)處理架構(gòu)Twitter是全球訪問量最大的社交網(wǎng)站之一，Twitter268.4.2Storm的特點Storm可用于許多領(lǐng)域中，如實時分析、在線機器學(xué)習(xí)、持續(xù)計算、遠程RPC、數(shù)據(jù)提取加載轉(zhuǎn)換等Storm具有以下主要特點：整合性：Storm可方便地與隊列系統(tǒng)和數(shù)據(jù)庫系統(tǒng)進行整合簡易的API：Storm的API在使用上即簡單又方便可擴展性：Storm的并行特性使其可以運行在分布式集群中容錯性：Storm可自動進行故障節(jié)點的重啟、任務(wù)的重新分配可靠的消息處理：Storm保證每個消息都能完整處理支持各種編程語言：Storm支持使用各種編程語言來定義任務(wù)快速部署：Storm可以快速進行部署和使用免費、開源：Storm是一款開源框架，可以免費使用8.4.2Storm的特點Storm可用于許多領(lǐng)域中，如實278.4.3Storm設(shè)計思想Storm主要術(shù)語包括Streams、Spouts、Bolts、Topology和StreamGroupingsStreams：Storm將流數(shù)據(jù)Stream描述成一個無限的Tuple序列，這些Tuple序列會以分布式的方式并行地創(chuàng)建和處理每個tuple是一堆值，每個值有一個名字，并且每個值可以是任何類型Tuple本來應(yīng)該是一個Key-Value的Map，由于各個組件間傳遞的tuple的字段名稱已經(jīng)事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個ValueList（值列表）Field1Field2Field3Field48.4.3Storm設(shè)計思想Storm主要術(shù)語包括Stre288.4.3Storm設(shè)計思想Spout：Storm認為每個Stream都有一個源頭，并把這個源頭抽象為Spout通常Spout會從外部數(shù)據(jù)源（隊列、數(shù)據(jù)庫等）讀取數(shù)據(jù)，然后封裝成Tuple形式，發(fā)送到Stream中。Spout是一個主動的角色，在接口內(nèi)部有個nextTuple函數(shù)，Storm框架會不停的調(diào)用該函數(shù)8.4.3Storm設(shè)計思想Spout：Storm認為每個298.4.3Storm設(shè)計思想Bolt：Storm將Streams的狀態(tài)轉(zhuǎn)換過程抽象為Bolt。Bolt即可以處理Tuple，也可以將處理后的Tuple作為新的Streams發(fā)送給其他BoltBolt可以執(zhí)行過濾、函數(shù)操作、Join、操作數(shù)據(jù)庫等任何操作Bolt是一個被動的角色，其接口中有一個execute(Tupleinput)方法，在接收到消息之后會調(diào)用此函數(shù)，用戶可以在此方法中執(zhí)行自己的處理邏輯8.4.3Storm設(shè)計思想Bolt：Storm將Stre308.4.3Storm設(shè)計思想Topology：Storm將Spouts和Bolts組成的網(wǎng)絡(luò)抽象成Topology，它可以被提交到Storm集群執(zhí)行。Topology可視為流轉(zhuǎn)換圖，圖中節(jié)點是一個Spout或Bolt，邊則表示Bolt訂閱了哪個Stream。當(dāng)Spout或者Bolt發(fā)送元組時，它會把元組發(fā)送到每個訂閱了該Stream的Bolt上進行處理Topology里面的每個處理組件（Spout或Bolt）都包含處理邏輯，而組件之間的連接則表示數(shù)據(jù)流動的方向Topology里面的每一個組件都是并行運行的在Topology里面可以指定每個組件的并行度，Storm會在集群里面分配那么多的線程來同時計算在Topology的具體實現(xiàn)上，Storm中的Topology定義僅僅是一些Thrift結(jié)構(gòu)體（二進制高性能的通信中間件），支持各種編程語言進行定義SpoutSpoutBoltBoltBoltBoltBolt8.4.3Storm設(shè)計思想Topology：Storm將318.4.3Storm設(shè)計思想StreamGroupings：Storm中的StreamGroupings用于告知Topology如何在兩個組件間（如Spout和Bolt之間，或者不同的Bolt之間）進行Tuple的傳送。每一個Spout和Bolt都可以有多個分布式任務(wù)，一個任務(wù)在什么時候、以什么方式發(fā)送Tuple就是由StreamGroupings來決定的8.4.3Storm設(shè)計思想StreamGrouping328.4.3Storm設(shè)計思想目前，Storm中的StreamGroupings有如下幾種方式：(1)ShuffleGrouping：隨機分組，隨機分發(fā)Stream中的Tuple，保證每個Bolt的Task接收Tuple數(shù)量大致一致(2)FieldsGrouping：按照字段分組，保證相同字段的Tuple分配到同一個Task中(3)AllGrouping：廣播發(fā)送，每一個Task都會收到所有的Tuple(4)GlobalGrouping：全局分組，所有的Tuple都發(fā)送到同一個Task中(5)NonGrouping：不分組，和ShuffleGrouping類似，當(dāng)前Task的執(zhí)行會和它的被訂閱者在同一個線程中執(zhí)行(6)DirectGrouping：直接分組，直接指定由某個Task來執(zhí)行Tuple的處理8.4.3Storm設(shè)計思想目前，Storm中的Strea338.4.4Storm框架設(shè)計HadoopStorm應(yīng)用名稱JobTopology系統(tǒng)角色JobTrackerNimbusTaskTrackerSupervisor組件接口Map/ReduceSpout/BoltStorm和Hadoop架構(gòu)組件功能對應(yīng)關(guān)系Storm運行任務(wù)的方式與Hadoop類似：Hadoop運行的是MapReduce作業(yè)，而Storm運行的是“Topology”但兩者的任務(wù)大不相同，主要的不同是：MapReduce作業(yè)最終會完成計算并結(jié)束運行，而Topology將持續(xù)處理消息（直到人為終止）8.4.4Storm框架設(shè)計HadoopStorm應(yīng)用名稱348.4.4Storm框架設(shè)計Storm集群采用“Master—Worker”的節(jié)點方式：Master節(jié)點運行名為“Nimbus”的后臺程序（類似Hadoop中的“JobTracker”），負責(zé)在集群范圍內(nèi)分發(fā)代碼、為Worker分配任務(wù)和監(jiān)測故障Worker節(jié)點運行名為“Supervisor”的后臺程序，負責(zé)監(jiān)聽分配給它所在機器的工作，即根據(jù)Nimbus分配的任務(wù)來決定啟動或停止Worker進程，一個Worker節(jié)點上同時運行若干個Worker進程8.4.4Storm框架設(shè)計Storm集群采用“Maste35Storm使用Zookeeper來作為分布式協(xié)調(diào)組件，負責(zé)Nimbus和多個Supervisor之間的所有協(xié)調(diào)工作。借助于Zookeeper，若Nimbus進程或Supervisor進程意外終止，重啟時也能讀取、恢復(fù)之前的狀態(tài)并繼續(xù)工作，使得Storm極其穩(wěn)定8.4.4Storm框架設(shè)計Storm集群架構(gòu)示意圖Worker進程Worker進程Worker進程……Storm使用Zookeeper來作為分布式協(xié)調(diào)組件，負責(zé)N368.4.4Storm框架設(shè)計(1)worker:每個worker進程都屬于一個特定的Topology，每個Supervisor節(jié)點的worker可以有多個，每個worker對Topology中的每個組件（Spout或Bolt）運行一個或者多個executor線程來提供task的運行服務(wù)(2)executor：executor是產(chǎn)生于worker進程內(nèi)部的線程，會執(zhí)行同一個組件的一個或者多個task。(3)task:實際的數(shù)據(jù)處理由task完成，在Topology的生命周期中，每個組件的task數(shù)目是不會發(fā)生變化的，而executor的數(shù)目卻不一定。executor數(shù)目小于等于task的數(shù)目，默認情況下，二者是相等的Worker、Executor和Task的關(guān)系8.4.4Storm框架設(shè)計(1)worker:每個wor378.4.4Storm框架設(shè)計基于這樣的架構(gòu)設(shè)計，Storm的工作流程如下圖所示：Storm工作流程示意圖所有Topology任務(wù)的提交必須在Storm客戶端節(jié)點上進行，提交后，由Nimbus節(jié)點分配給其他Supervisor節(jié)點進行處理Nimbus節(jié)點首先將提交的Topology進行分片，分成一個個Task，分配給相應(yīng)的Supervisor，并將Task和Supervisor相關(guān)的信息提交到Zookeeper集群上Supervisor會去Zookeeper集群上認領(lǐng)自己的Task，通知自己的Worker進程進行Task的處理說明：在提交了一個Topology之后，Storm就會創(chuàng)建Spout/Bolt實例并進行序列化。之后，將序列化的組件發(fā)送給所有的任務(wù)所在的機器(即Supervisor節(jié)點)，在每一個任務(wù)上反序列化組件8.4.4Storm框架設(shè)計基于這樣的架構(gòu)設(shè)計，Storm388.5SparkStreaming8.5.1SparkStreaming設(shè)計8.5.2SparkStreaming與Storm的對比8.5SparkStreaming8.5.1Spark398.5.1SparkStreaming設(shè)計SparkStreaming可整合多種輸入數(shù)據(jù)源，如Kafka、Flume、HDFS，甚至是普通的TCP套接字。經(jīng)處理后的數(shù)據(jù)可存儲至文件系統(tǒng)、數(shù)據(jù)庫，或顯示在儀表盤里圖13SparkStreaming支持的輸入、輸出數(shù)據(jù)源8.5.1SparkStreaming設(shè)計SparkS408.5.1SparkStreaming設(shè)計SparkStreaming的基本原理是將實時輸入數(shù)據(jù)流以時間片（秒級）為單位進行拆分，然后經(jīng)Spark引擎以類似批處理的方式處理每個時間片數(shù)據(jù)圖14SparkStreaming執(zhí)行流程8.5.1SparkStreaming設(shè)計SparkS418.5.1SparkStreaming設(shè)計SparkStreaming最主要的抽象是DStream（DiscretizedStream，離散化數(shù)據(jù)流），表示連續(xù)不斷的數(shù)據(jù)流。在內(nèi)部實現(xiàn)上，SparkStreaming的輸入數(shù)據(jù)按照時間片（如1秒）分成一段一段的DStream，每一段數(shù)據(jù)轉(zhuǎn)換為Spark中的RDD，并且對DStream的操作都最終轉(zhuǎn)變?yōu)閷ο鄳?yīng)的RDD的操作圖15DStream操作示意圖8.5.1SparkStreaming設(shè)計SparkS428.5.2SparkStreaming與Storm的對比SparkStreaming和Storm最大的區(qū)別在于，SparkStreaming無法實現(xiàn)毫秒級的流計算，而Storm可以實現(xiàn)毫秒級響應(yīng)SparkStreaming構(gòu)建在Spark上，一方面是因為Spark的低延遲執(zhí)行引擎（100ms+）可以用于實時計算，另一方面，相比于Storm，RDD數(shù)據(jù)集更容易做高效的容錯處理SparkStreaming采用的小批量處理的方式使得它可以同時兼容批量和實時數(shù)據(jù)處理的邏輯和算法，因此，方便了一些需要歷史數(shù)據(jù)和實時數(shù)據(jù)聯(lián)合分析的特定應(yīng)用場合8.5.2SparkStreaming與Storm的對比438.6Samza8.6.1基本概念8.6.2系統(tǒng)架構(gòu)8.6Samza8.6.1基本概念448.6.1基本概念1.作業(yè)一個作業(yè)（Job）是對一組輸入流進行處理轉(zhuǎn)化成輸出流的程序。8.6.1基本概念1.作業(yè)458.6.1基本概念2.分區(qū)Samza的流數(shù)據(jù)單位既不是Storm中的元組，也不是SparkStreaming中的DStream，而是一條條消息Samza中的每個流都被分割成一個或多個分區(qū)，對于流里的每一個分區(qū)而言，都是一個有序的消息序列，后續(xù)到達的消息會根據(jù)一定規(guī)則被追加到其中一個分區(qū)里8.6.1基本概念2.分區(qū)Samza的流數(shù)據(jù)單位既不是St468.6.1基本概念3.任務(wù)一個作業(yè)會被進一步分割成多個任務(wù)（Task）來執(zhí)行，其中，每個任務(wù)負責(zé)處理作業(yè)中的一個分區(qū)分區(qū)之間沒有定義順序，從而允許每一個任務(wù)獨立執(zhí)行YARN調(diào)度器負責(zé)把任務(wù)分發(fā)給各個機器，最終，一個工作中的多個任務(wù)會被分發(fā)到多個機器進行分布式并行處理8.6.1基本概念3.任務(wù)478.6.1基本概念4.數(shù)據(jù)流圖一個數(shù)據(jù)流圖是由多個作業(yè)構(gòu)成的，其中，圖中的每個節(jié)點表示包含數(shù)據(jù)的流，每條邊表示數(shù)據(jù)傳輸多個作業(yè)串聯(lián)起來就完成了流式的數(shù)據(jù)處理流程由于采用了異步的消息訂閱分發(fā)機制，不同任務(wù)之間可以獨立運行8.6.1基本概念4.數(shù)據(jù)流圖一個數(shù)據(jù)流圖是由多個作業(yè)構(gòu)成488.6.2系統(tǒng)架構(gòu)MapReduce批處理架構(gòu)Samza流處理架構(gòu)數(shù)據(jù)層HDFSKafka執(zhí)行層YARNYARN處理層MapReduceSamzaAPI表

MapReduce批處理架構(gòu)和Samza流處理架構(gòu)的類比Samza系統(tǒng)架構(gòu)主要包括流數(shù)據(jù)層（Kafka）執(zhí)行層（YARN）處理層（SamzaAPI）流處理層和執(zhí)行層都被設(shè)計成可插拔的，開發(fā)人員可以使用其他框架來替代YARN和Kafka8.6.2系統(tǒng)架構(gòu)MapReduce批處理架構(gòu)Samza流498.6.2系統(tǒng)架構(gòu)處理分析過程如下：Samza客戶端需要執(zhí)行一個Samza作業(yè)時，它會向YARN的ResouceManager提交作業(yè)請求ResouceManager通過與NodeManager溝通為該作業(yè)分配容器（包含了CPU、內(nèi)存等資源）來運行SamzaApplicationMasterSamzaApplicationMaster進一步向ResourceManager申請運行任務(wù)的容器獲得容器后，SamzaApplicationMaster與容器所在的NodeManager溝通，啟動該容器，并在其中運行SamzaTaskRunnerSamzaTaskRunner負責(zé)執(zhí)行具體的Samza任務(wù)，完成流數(shù)據(jù)處理分析8.6.2系統(tǒng)架構(gòu)處理分析過程如下：508.7Storm、SparkStreaming和Samza的應(yīng)用場景從編程的靈活性來講，Storm是比較理想的選擇，它使用ApacheThrift，可以用任何編程語言來編寫拓撲結(jié)構(gòu)（Topology）當(dāng)需要在一個集群中把流計算和圖計算、機器學(xué)習(xí)、SQL查詢分析等進行結(jié)合時，可以選擇SparkStreaming，因為，在Spark上可以統(tǒng)一部署SparkSQL，SparkStreaming、MLlib，GraphX等組件，提供便捷的一體化編程模型當(dāng)有大量的狀態(tài)需要處理時，比如每個分區(qū)都有數(shù)十億個元組，則可以選擇Samza。當(dāng)應(yīng)用場景需要毫秒級響應(yīng)時，可以選擇Storm和Samza，因為SparkStreaming無法實現(xiàn)毫秒級的流計算8.7Storm、SparkStreaming和Samz518.8Storm編程實踐8.8.1編寫Storm程序8.8.2安裝Storm的基本過程8.8.3運行Storm程序Storm上機實踐詳細過程，請參考廈門大學(xué)數(shù)據(jù)庫實驗室建設(shè)的“中國高校大數(shù)據(jù)課程公共服務(wù)平臺”中的“大數(shù)據(jù)課程學(xué)生服務(wù)站”中的“學(xué)習(xí)指南”欄目：學(xué)生服務(wù)站地址：/post/4331/學(xué)習(xí)指南欄目中包含了《第八章流計算學(xué)習(xí)指南》/blog/767/掃一掃訪問學(xué)生服務(wù)站8.8Storm編程實踐8.8.1編寫Storm程序St528.8.1編寫Storm程序基于Storm的單詞統(tǒng)計在形式上與基于MapReduce的單詞統(tǒng)計是類似的，MapReduce使用的是Map和Reduce的抽象，而Storm使用的是Spout和Bolt的抽象Storm進行單詞統(tǒng)計的整個流程：從Spout中發(fā)送Stream（每個英文句子為一個Tuple）用于分割單詞的Bolt將接收的句子分解為獨立的單詞，將單詞作為Tuple的字段名發(fā)送出去用于計數(shù)的Bolt接收表示單詞的Tuple，并對其進行統(tǒng)計輸出每個單詞以及單詞出現(xiàn)過的次數(shù)程序任務(wù)：單詞統(tǒng)計8.8.1編寫Storm程序基于Storm的單詞統(tǒng)計在形式538.8.1編寫Storm程序一個句子經(jīng)Storm的單詞統(tǒng)計得出的結(jié)果8.8.1編寫Storm程序一個句子經(jīng)Storm的單詞統(tǒng)計54importorg.apache.storm.Config;Import……publicclassWordCountTopology{publicstaticclassRandomSentenceSpoutextendsBaseRichSpout{……}publicstaticclassSplitSentenceextendsShellBoltimplementsIRichBolt{……}publicstaticclassWordCountextendsBaseBasicBolt{……}publicstaticvoidmain(String[]args)throwsException{TopologyBuilderbuilder=newTopologyBuilder();builder.setSpout("sentences",newRandomSentenceSpout(),5);builder.setBolt("split",newSplitSentence(),8).shuffleGrouping("sentences");builder.setBolt("count",newWordCount(),12).fieldsGrouping("split",newFields("word"));……}}8.8.1編寫Storm程序Storm的編程模型非常簡單，如下代碼即定義了整個單詞統(tǒng)計Topology的整體邏輯importorg.apache.storm.Config558.8.1編寫Storm程序main()函數(shù)中的處理邏輯Topology中僅定義了整體的計算邏輯，還需要定義具體的處理函數(shù)。具體的處理函數(shù)可以使用任一編程語言來定義，甚至也可以結(jié)合多種編程語言來實現(xiàn)8.8.1編寫Storm程序main()函數(shù)中的處理邏輯T56publicclassRandomSentenceSpoutextendsBaseRichSpout{SpoutOutputCollector_collector;Random_rand;@OverridepublicvoidnextTuple(){Utils.sleep(100);String[]sentences=newString[]{"thecowjumpedoverthemoon","anappleadaykeepsthedoctoraway","fourscoreandsevenyearsago","snowwhiteandthesevendwarfs","iamattwowithnature"};Stringsentence=sentences[_rand.nextInt(sentences.length)];_collector.emit(newValues(sentence));}@OverridepublicvoiddeclareOutputFields(OutputFieldsDeclarerdeclarer){declarer.declare(newFields(“sentences"));}}8.8.1編寫Storm程序備注：

為簡單起見，RandomSentenceSpout省略了類中的一些方法RandomSentenceSpout類publicclassRandomSentenceSpo578.8.1編寫Storm程序如SplitSentence()方法雖然是通過Java語言定義的，但具體的操作可通過Python腳本來完成Topology里面的每個組件必須定義它要發(fā)射的Tuple的每個字段SplitSentence類8.8.1編寫Storm程序如SplitSentence(588.8.1編寫Storm程序Python腳本splitsentence.py定義了一個簡單的單詞分割方法，即通過空格來分割單詞。分割后的單詞通過emit()方法以Tuple的形式發(fā)送給訂閱了該Stream的Bolt進行接收和處理splitsentence.py8.8.1編寫Storm程序Python腳本splitse598.8.1編寫Storm程序單詞統(tǒng)計的具體邏輯：首先判斷單詞是否統(tǒng)計過，若未統(tǒng)計過，需先將count值置為0。若單詞已統(tǒng)計過，則每出現(xiàn)一次該單詞，count值就加1WordCount類

8.8.1編寫Storm程序單詞統(tǒng)計的具體邏輯：首先判斷單608.8.1編寫Storm程序上述雖然是一個簡單的單詞統(tǒng)計，但對其進行擴展，便可應(yīng)用到許多場景中，如微博中的實時熱門話題。Twitter也正是使用了Storm框架實現(xiàn)了實時熱門話題Twitter實時熱門話題處理流程示意圖8.8.1編寫Storm程序上述雖然是一個簡單的單詞統(tǒng)計，618.8.2安裝Storm的基本過程安裝Storm的基本過程如下：第一步：安裝Java環(huán)境第二步：安裝Zookeeper第三步：安裝Storm（單機）第四步：關(guān)閉Storm本實例中Storm具體運行環(huán)境如下：CentOS6.4Storm0.9.6JavaJDK1.7ZooKeeper3.4.6Python2.6備注：CentOS中已默認安裝了Python2.6，我們還需要安裝JDK環(huán)境以及分布式應(yīng)用程序協(xié)調(diào)服務(wù)ZookeeperStorm上機實踐詳細過程，請參考廈門大學(xué)數(shù)據(jù)庫實驗室建設(shè)的“中國高校大數(shù)據(jù)課程公共服務(wù)平臺”中的“大數(shù)據(jù)課程學(xué)生服務(wù)站”中的“學(xué)習(xí)指南”欄目：學(xué)生服務(wù)站地址：/post/4331/學(xué)習(xí)指南欄目中包含了《第八章流計算學(xué)習(xí)指南》/blog/767/8.8.2安裝Storm的基本過程安裝Storm的基本過程628.8.2安裝Storm的基本過程第一步：安裝Java環(huán)境Storm運行需要Java環(huán)境，可選擇Oracle的JDK，或是OpenJDK，現(xiàn)在一般Linux系統(tǒng)默認安裝的基本是OpenJDK，如CentOS6.4就默認安裝了OpenJDK1.7。但需要注意的是，CentOS6.4中默認安裝的只是JavaJRE，而不是JDK，為了開發(fā)方便，我們還是需要通過yum進行安裝JDK$sudoyuminstalljava-1.7.0-openjdkjava-1.7.0-openjdk-devel接著需要配置一下JAVA_HOME環(huán)境變量，為方便，可以在~/.bashrc中進行設(shè)置8.8.2安裝Storm的基本過程第一步：安裝Java環(huán)境638.8.2安裝Storm的基本過程第二步：安裝Zookeeper到官網(wǎng)下載Zookeeper，比如下載

“zookeeper-3.4.6.tar.gz”

下載后執(zhí)行如下命令進行安裝zookeeper（將命令中3.4.6改為你下載的版本）：$sudotar-zxf~/下載/zookeeper-3.4.6.tar.gz-C/usr/local$cd/usr/local$sudomvzookeeper-*zookeeper#修改目錄名稱方便使用$sudochown-Rhadoop:hadoop./zookeeper#此處的hadoop為你的用戶名chown命令讓hadoop用戶擁有zookeeper目錄下的所有文件的權(quán)限8.8.2安裝Storm的基本過程第二步：安裝Zookee648.8.2安裝Storm的基本過程接著執(zhí)行如下命令進行zookeeper配置：$cd/usr/local/zookeeper$mkdirtmp$cp./conf/zoo_sample.cfg./conf/zoo.cfg$vim./conf/zoo.cfg進入zoo.cfg文件編輯狀態(tài)后，將當(dāng)中的dataDir=/tmp/zookeeper更改為dataDir=/usr/local/zookeeper/tmp。接著執(zhí)行：$./bin/zkServer.shstart第二步：安裝Zookeeper（續(xù)）8.8.2安裝Storm的基本過程接著執(zhí)行如下命令進行zo658.8.2安裝Storm的基本過程第三步：安裝Storm（單機）到官網(wǎng)下載Storm，比如Storm0.9.6下載后執(zhí)行如下命令進行安裝Storm：$sudotar-zxf~/下載/apache-storm-0.9.6.tar.gz-C/usr/local$cd/usr/local$sudomvapache-storm-0.9.6storm$sudochown-Rhadoop:hadoop./storm#此處的hadoop為你的用戶名接著執(zhí)行如下命令進行Storm配置:$cd/usr/local/storm$vim./conf/storm.yaml備注：storm的運行有兩種模式:本地模式和分布式模式.在本地模式中，storm用一個進程里面的線程來模擬所有的spout和bolt.本地模式對開發(fā)和測試來說比較有用。在分布式模式下，storm由一堆機器組成。當(dāng)提交topology給master的時候，master負責(zé)分發(fā)代碼并且負責(zé)給topolgoy分配工作進程。如果一個工作進程掛掉了，master節(jié)點會把它重新分配到其它節(jié)點8.8.2安裝Storm的基本過程第三步：安裝Storm（668.8.2安裝Storm的基本過程修改其中的storm.zookeeper.servers和nimbus.host兩個配置項，即取消掉注釋且都修改值為（我們只需要在單機上運行），如下圖所示。第三步：安裝Storm（單機）（續(xù)）然后就可以啟動Storm了。執(zhí)行如下命令啟動nimbus后臺進程：$./bin/stormnimbus8.8.2安裝Storm的基本過程修改其中的storm.678.8.2安裝Storm的基本過程啟動nimbus后，終端被該進程占用了，不能再繼續(xù)執(zhí)行其他命令了。因此我們需要另外開啟一個終端，然后執(zhí)行如下命令啟動supervisor后臺進程：第三步：安裝Storm（單機）（續(xù)）$#需要另外開啟一個終端$/usr/local/storm/bin/stormsupervisor同樣的，啟動supervisor后，我們還需要開啟另外的終端才能執(zhí)行其他命令。另外，我們可以使用

jps

命令檢查是否成功啟動，若成功啟動會顯示nimbus、supervisor、QuorumPeeMain（QuorumPeeMain是zookeeper的后臺進程，若顯示config_value表明nimbus或supervisor還在啟動中），如下圖所示。8.8.2安裝Storm的基本過程啟動nimbus后，688.8.2安裝Storm的基本過程第四步：關(guān)閉Storm之前啟動的nimbus和supervisor占用了兩個終端窗口，切換到這兩個終端窗口，按鍵盤的Ctrl+C可以終止進程，終止后，也就相當(dāng)于關(guān)閉了Storm。8.8.2安裝Storm的基本過程第四步：關(guān)閉Storm之69運行Storm計算任務(wù)，就是提交Topology運行一個Topology是很簡單的。首先，把所有的代碼以及所依賴的jar打進一個jar包。然后運行類似下面的這個命令stromjarall-your-code.jarbacktype.storm.MyTopologyarg1arg2stormjar負責(zé)連接到nimbus并且上傳jar文件8.8.3運行Storm實例運行Storm計算任務(wù)，就是提交Topology8.8.3708.8.3運行Storm實例Storm中自帶了一些例子，我們可以執(zhí)行一下WordCount例子來感受一下Storm的執(zhí)行流程。執(zhí)行如下命令：$/usr/local/storm/bin/stormjar/usr/local/storm/examples/storm-starter/storm-starter-topologies-0.9.6.jarstorm.starter.WordCountTopology該程序是不斷地取如下四句英文句子中的一句作為數(shù)據(jù)源，然后發(fā)送給bolt來統(tǒng)計單詞出現(xiàn)的次數(shù)。{"thecowjumpedoverthemoon","anappleadaykeepsthedoctoraway","fourscoreandsevenyearsago","snowwhiteandthesevendwarfs","iamattwowithnature"}

8.8.3運行Storm實例Storm中自帶了一些例子，我712022/12/17商業(yè)大數(shù)據(jù)分析2022/12/17商業(yè)大數(shù)據(jù)分析72提綱8.1流計算概述8.2流計算處理流程8.3流計算應(yīng)用8.4流計算開源框架–Storm8.5SparkStreaming8.6Samza8.7Storm、SparkStreaming和Samza的應(yīng)用場景8.8Storm編程實踐提綱8.1流計算概述738.1 流計算概述8.1.1 靜態(tài)數(shù)據(jù)和流數(shù)據(jù)8.1.2 批量計算和實時計算8.1.3 流計算概念8.1.4 流計算與Hadoop8.1.5 流計算框架8.1 流計算概述8.1.1 靜態(tài)數(shù)據(jù)和流數(shù)據(jù)74很多企業(yè)為了支持決策分析而構(gòu)建的數(shù)據(jù)倉庫系統(tǒng)，其中存放的大量歷史數(shù)據(jù)就是靜態(tài)數(shù)據(jù)。技術(shù)人員可以利用數(shù)據(jù)挖掘和OLAP（On-LineAnalyticalProcessing）分析工具從靜態(tài)數(shù)據(jù)中找到對企業(yè)有價值的信息8.1.1靜態(tài)數(shù)據(jù)和流數(shù)據(jù)很多企業(yè)為了支持決策分析而構(gòu)建的數(shù)據(jù)倉庫系統(tǒng)，其中存放的大量75近年來，在Web應(yīng)用、網(wǎng)絡(luò)監(jiān)控、傳感監(jiān)測等領(lǐng)域，興起了一種新的數(shù)據(jù)密集型應(yīng)用——流數(shù)據(jù)，即數(shù)據(jù)以大量、快速、時變的流形式持續(xù)到達實例：PM2.5檢測、電子商務(wù)網(wǎng)站用戶點擊流流數(shù)據(jù)具有如下特征：數(shù)據(jù)快速持續(xù)到達，潛在大小也許是無窮無盡的數(shù)據(jù)來源眾多，格式復(fù)雜數(shù)據(jù)量大，但是不十分關(guān)注存儲，一旦經(jīng)過處理，要么被丟棄，要么被歸檔存儲注重數(shù)據(jù)的整體價值，不過分關(guān)注個別數(shù)據(jù)數(shù)據(jù)順序顛倒，或者不完整，系統(tǒng)無法控制將要處理的新到達的數(shù)據(jù)元素的順序8.1.1靜態(tài)數(shù)據(jù)和流數(shù)據(jù)近年來，在Web應(yīng)用、網(wǎng)絡(luò)監(jiān)控、傳感監(jiān)測等領(lǐng)域，興起了一種新76對靜態(tài)數(shù)據(jù)和流數(shù)據(jù)的處理，對應(yīng)著兩種截然不同的計算模式：批量計算和實時計算8.1.2批量計算和實時計算圖8-2數(shù)據(jù)的兩種處理模型批量計算：充裕時間處理靜態(tài)數(shù)據(jù)，如Hadoop流數(shù)據(jù)不適合采用批量計算，因為流數(shù)據(jù)不適合用傳統(tǒng)的關(guān)系模型建模流數(shù)據(jù)必須采用實時計算，響應(yīng)時間為秒級數(shù)據(jù)量少時，不是問題，但是，在大數(shù)據(jù)時代，數(shù)據(jù)格式復(fù)雜、來源眾多、數(shù)據(jù)量巨大，對實時計算提出了很大的挑戰(zhàn)。因此，針對流數(shù)據(jù)的實時計算——流計算，應(yīng)運而生對靜態(tài)數(shù)據(jù)和流數(shù)據(jù)的處理，對應(yīng)著兩種截然不同的計算模式：批量77流計算：實時獲取來自不同數(shù)據(jù)源的海量數(shù)據(jù)，經(jīng)過實時分析處理，獲得有價值的信息8.1.3流計算概念圖8-3流計算示意圖流計算：實時獲取來自不同數(shù)據(jù)源的海量數(shù)據(jù)，經(jīng)過實時分析處理，78流計算秉承一個基本理念，即數(shù)據(jù)的價值隨著時間的流逝而降低，如用戶點擊流。因此，當(dāng)事件出現(xiàn)時就應(yīng)該立即進行處理，而不是緩存起來進行批量處理。為了及時處理流數(shù)據(jù)，就需要一個低延遲、可擴展、高可靠的處理引擎對于一個流計算系統(tǒng)來說，它應(yīng)達到如下需求：高性能：處理大數(shù)據(jù)的基本要求，如每秒處理幾十萬條數(shù)據(jù)海量式：支持TB級甚至是PB級的數(shù)據(jù)規(guī)模實時性：保證較低的延遲時間，達到秒級別，甚至是毫秒級別分布式：支持大數(shù)據(jù)的基本架構(gòu)，必須能夠平滑擴展易用性：能夠快速進行開發(fā)和部署可靠性：能可靠地處理流數(shù)據(jù)8.1.3流計算概念流計算秉承一個基本理念，即數(shù)據(jù)的價值隨著時間的流逝而降低，如79Hadoop設(shè)計的初衷是面向大規(guī)模數(shù)據(jù)的批量處理，每臺機器并行運行MapReduce任務(wù)，最后對結(jié)果進行匯總輸出MapReduce是專門面向靜態(tài)數(shù)據(jù)的批量處理的，內(nèi)部各種實現(xiàn)機制都為批處理做了高度優(yōu)化，不適合用于處理持續(xù)到達的動態(tài)數(shù)據(jù)可能會想到一種“變通”的方案來降低批處理的時間延遲——將基于MapReduce的批量處理轉(zhuǎn)為小批量處理，將輸入數(shù)據(jù)切成小的片段，每隔一個周期就啟動一次MapReduce作業(yè)。但這種方式也無法有效處理流數(shù)據(jù)切分成小片段，可以降低延遲，但是也增加了附加開銷，還要處理片段之間依賴關(guān)系需要改造MapReduce以支持流式處理8.1.4流計算與Hadoop結(jié)論：魚和熊掌不可兼得，Hadoop擅長批處理，不適合流計算Hadoop設(shè)計的初衷是面向大規(guī)模數(shù)據(jù)的批量處理，每臺機器并80當(dāng)前業(yè)界誕生了許多專門的流數(shù)據(jù)實時計算系統(tǒng)來滿足各自需求目前有三類常見的流計算框架和平臺：商業(yè)級的流計算平臺、開源流計算框架、公司為支持自身業(yè)務(wù)開發(fā)的流計算框架商業(yè)級：IBMInfoSphereStreams和IBMStreamBase較為常見的是開源流計算框架，代表如下：TwitterStorm：免費、開源的分布式實時計算系統(tǒng)，可簡單、高效、可靠地處理大量的流數(shù)據(jù)Yahoo!S4（SimpleScalableStreamingSystem）：開源流計算平臺，是通用的、分布式的、可擴展的、分區(qū)容錯的、可插拔的流式系統(tǒng)公司為支持自身業(yè)務(wù)開發(fā)的流計算框架：FacebookPumaDstream（百度）銀河流數(shù)據(jù)處理平臺（淘寶）8.1.5流計算框架當(dāng)前業(yè)界誕生了許多專門的流數(shù)據(jù)實時計算系統(tǒng)來滿足各自需求8.818.2 流計算處理流程8.2.1 概述8.2.2 數(shù)據(jù)實時采集8.2.3 數(shù)據(jù)實時計算8.2.4 實時查詢服務(wù)8.2 流計算處理流程8.2.1 概述82傳統(tǒng)的數(shù)據(jù)處理流程，需要先采集數(shù)據(jù)并存儲在關(guān)系數(shù)據(jù)庫等數(shù)據(jù)管理系統(tǒng)中，之后由用戶通過查詢操作和數(shù)據(jù)管理系統(tǒng)進行交互傳統(tǒng)的數(shù)據(jù)處理流程隱含了兩個前提：存儲的數(shù)據(jù)是舊的。存儲的靜態(tài)數(shù)據(jù)是過去某一時刻的快照，這些數(shù)據(jù)在查詢時可能已不具備時效性了需要用戶主動發(fā)出查詢來獲取結(jié)果8.2.1數(shù)據(jù)處理流程傳統(tǒng)的數(shù)據(jù)處理流程示意圖傳統(tǒng)的數(shù)據(jù)處理流程，需要先采集數(shù)據(jù)并存儲在關(guān)系數(shù)據(jù)庫等數(shù)據(jù)管83流計算的處理流程一般包含三個階段：數(shù)據(jù)實時采集、數(shù)據(jù)實時計算、實時查詢服務(wù)8.2.1數(shù)據(jù)處理流程流計算處理流程示意圖流計算的處理流程一般包含三個階段：數(shù)據(jù)實時采集、數(shù)據(jù)實時計算84數(shù)據(jù)實時采集階段通常采集多個數(shù)據(jù)源的海量數(shù)據(jù)，需要保證實時性、低延遲與穩(wěn)定可靠以日志數(shù)據(jù)為例，由于分布式集群的廣泛應(yīng)用，數(shù)據(jù)分散存儲在不同的機器上，因此需要實時匯總來自不同機器上的日志數(shù)據(jù)目前有許多互聯(lián)網(wǎng)公司發(fā)布的開源分布式日志采集系統(tǒng)均可滿足每秒數(shù)百MB的數(shù)據(jù)采集和傳輸需求，如：Facebook的ScribeLinkedIn的Kafka淘寶的TimeTunnel基于Hadoop的Chukwa和Flume8.2.2數(shù)據(jù)實時采集數(shù)據(jù)實時采集階段通常采集多個數(shù)據(jù)源的海量數(shù)據(jù)，需要保證實時性85數(shù)據(jù)采集系統(tǒng)的基本架構(gòu)一般有以下三個部分：Agent：主動采集數(shù)據(jù)，并把數(shù)據(jù)推送到Collector部分Collector：接收多個Agent的數(shù)據(jù)，并實現(xiàn)有序、可靠、高性能的轉(zhuǎn)發(fā)Store：存儲Collector轉(zhuǎn)發(fā)過來的數(shù)據(jù)（對于流計算不存儲數(shù)據(jù)）8.2.2數(shù)據(jù)實時采集數(shù)據(jù)采集系統(tǒng)基本架構(gòu)數(shù)據(jù)采集系統(tǒng)的基本架構(gòu)一般有以下三個部分：8.2.2數(shù)據(jù)實86數(shù)據(jù)實時計算階段對采集的數(shù)據(jù)進行實時的分析和計算，并反饋實時結(jié)果經(jīng)流處理系統(tǒng)處理后的數(shù)據(jù)，可視情況進行存儲，以便之后再進行分析計算。在時效性要求較高的場景中，處理之后的數(shù)據(jù)也可以直接丟棄8.2.3數(shù)據(jù)實時計算數(shù)據(jù)實時計算流程數(shù)據(jù)實時計算階段對采集的數(shù)據(jù)進行實時的分析和計算，并反饋實時87實時查詢服務(wù)：經(jīng)由流計算框架得出的結(jié)果可供用戶進行實時查詢、展示或儲存?zhèn)鹘y(tǒng)的數(shù)據(jù)處理流程，用戶需要主動發(fā)出查詢才能獲得想要的結(jié)果。而在流處理流程中，實時查詢服務(wù)可以不斷更新結(jié)果，并將用戶所需的結(jié)果實時推送給用戶雖然通過對傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)進行定時查詢，也可以實現(xiàn)不斷地更新結(jié)果和結(jié)果推送，但通過這樣的方式獲取的結(jié)果，仍然是根據(jù)過去某一時刻的數(shù)據(jù)得到的結(jié)果，與實時結(jié)果有著本質(zhì)的區(qū)別8.2.3實時查詢服務(wù)實時查詢服務(wù)：經(jīng)由流計算框架得出的結(jié)果可供用戶進行實時查詢、88可見，流處理系統(tǒng)與傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)有如下不同：流處理系統(tǒng)處理的是實時的數(shù)據(jù)，而傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)處理的是預(yù)先存儲好的靜態(tài)數(shù)據(jù)用戶通過流處理系統(tǒng)獲取的是實時結(jié)果，而通過傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)，獲取的是過去某一時刻的結(jié)果流處理系統(tǒng)無需用戶主動發(fā)出查詢，實時查詢服務(wù)可以主動將實時結(jié)果推送給用戶8.2.3實時查詢服務(wù)可見，流處理系統(tǒng)與傳統(tǒng)的數(shù)據(jù)處理系統(tǒng)有如下不同：8.2.389流計算是針對流數(shù)據(jù)的實時計算，可以應(yīng)用在多種場景中，如Web服務(wù)、機器翻譯、廣告投放、自然語言處理、氣候模擬預(yù)測等如百度、淘寶等大型網(wǎng)站中，每天都會產(chǎn)生大量流數(shù)據(jù)，包括用戶的搜索內(nèi)容、用戶的瀏覽記錄等數(shù)據(jù)。采用流計算進行實時數(shù)據(jù)分析，可以了解每個時刻的流量變化情況，甚至可以分析用戶的實時瀏覽軌跡，從而進行實時個性化內(nèi)容推薦但是，并不是每個應(yīng)用場景都需要用到流計算的。流計算適合于需要處理持續(xù)到達的流數(shù)據(jù)、對數(shù)據(jù)處理有較高實時性要求的場景8.3流計算的應(yīng)用流計算是針對流數(shù)據(jù)的實時計算，可以應(yīng)用在多種場景中，如Web90傳統(tǒng)的業(yè)務(wù)分析一般采用分布式離線計算的方式，即將數(shù)據(jù)全部保存起來，然后每隔一定的時間進行離線分析來得到結(jié)果。但這樣會導(dǎo)致一定的延時，難以保證結(jié)果的實時性隨著分析業(yè)務(wù)對實時性要求的提升，離線分析模式已經(jīng)不適合用于流數(shù)據(jù)的分析，也不適用于要求實時響應(yīng)的互聯(lián)網(wǎng)應(yīng)用場景如淘寶網(wǎng)“雙十一”、“雙十二”的促銷活動，商家需要根據(jù)廣告效果來即時調(diào)整廣告，這就需要對廣告的受訪情況進行分析。但以往采用分布式離線分析，需要幾小時甚至一天的延時才能得到分析結(jié)果。而促銷活動只持續(xù)一天，因此，隔天才能得到的分析結(jié)果便失去了價值雖然分布式離線分析帶來的小時級的分析延時可以滿足大部分商家的需求，但隨著實時性要求越來越高，如何實現(xiàn)秒級別的實時分析響應(yīng)成為業(yè)務(wù)分析的一大挑戰(zhàn)8.3.1應(yīng)用場景1:實時分析傳統(tǒng)的業(yè)務(wù)分析一般采用分布式離線計算的方式，即將數(shù)據(jù)全部保存91針對流數(shù)據(jù)，“量子恒道”開發(fā)了海量數(shù)據(jù)實時流計算框架SuperMario。通過該框架，量子恒道可處理每天TB級的實時流數(shù)據(jù)，并且從用戶發(fā)出請求到數(shù)據(jù)展示，整個延時控制在2-3秒內(nèi)，達到了實時性的要求8.3.1應(yīng)用場景1:實時分析SuperMario處理流程針對流數(shù)據(jù)，“量子恒道”開發(fā)了海量數(shù)據(jù)實時流計算框架Supe92流計算不僅為互聯(lián)網(wǎng)帶來改變，也能改變我們的生活如提供導(dǎo)航路線，一般的導(dǎo)航路線并沒有考慮實時的交通狀況，即便在計算路線時有考慮交通狀況，往往也只是使用了以往的交通狀況數(shù)據(jù)。要達到根據(jù)實時交通狀態(tài)進行導(dǎo)航的效果，就需要獲取海量的實時交通數(shù)據(jù)并進行實時分析借助于流計算的實時特性，不僅可以根據(jù)交通情況制定路線，而且在行駛過程中，也可以根據(jù)交通情況的變化實時更新路線，始終為用戶提供最佳的行駛路線8.3.1應(yīng)用場景2:實時交通流計算不僅為互聯(lián)網(wǎng)帶來改變，也能改變我們的生活8.3.1應(yīng)938.4.1 Storm簡介8.4.2 Storm的特點8.4.3 Storm設(shè)計思想8.4.4

Storm框架設(shè)計8.4開源流計算框架Storm8.4.1 Storm簡介8.4開源流計算框架Storm94以前只有政府機構(gòu)和金融機構(gòu)能夠通過昂貴的定制系統(tǒng)來滿足流數(shù)據(jù)實時分析計算需求早期對于流計算的研究多數(shù)是基于對傳統(tǒng)數(shù)據(jù)庫處理的流式化，即實時數(shù)據(jù)庫，很少研究流計算框架Yahoo!S4和TwitterStorm的開源，改變了這個情況在流數(shù)據(jù)處理上比MapReduce更有優(yōu)勢批處理系統(tǒng)關(guān)注吞吐率，流處理系統(tǒng)關(guān)注延時Yahoo!S4和TwitterStorm改變了開發(fā)實時應(yīng)用的方式以前既要關(guān)注處理邏輯，還要解決實時數(shù)據(jù)獲取、傳輸、存儲現(xiàn)在可以快速低成本搭建起實時流處理系統(tǒng)8.4開源流計算框架Storm以前只有政府機構(gòu)和金融機構(gòu)能夠通過昂貴的定制系統(tǒng)來滿足流數(shù)據(jù)95TwitterStorm是一個免費、開源的分布式實時計算系統(tǒng)，Storm對于實時計算的意義類似于Hadoop對于批處理的意義，Storm可以簡單、高效、可靠地處理流數(shù)據(jù)，并支持多種編程語言Storm框架可以方便地與數(shù)據(jù)庫系統(tǒng)進行整合，從而開發(fā)出強大的實時計算系統(tǒng)8.4.1Storm簡介TwitterStorm是一個免費、開源的分布式實時計算系96Twitter是全球訪問量最大的社交網(wǎng)站之一，Twitter開發(fā)Storm流處理框架也是為了應(yīng)對其不斷增長的流數(shù)據(jù)實時處理需求8.4.1Storm簡介Twitter的分層數(shù)據(jù)處理架構(gòu)Twitter是全球訪問量最大的社交網(wǎng)站之一，Twitter978.4.2Storm的特點Storm可用于許多領(lǐng)域中，如實時分析、在線機器學(xué)習(xí)、持續(xù)計算、遠程RPC、數(shù)據(jù)提取加載轉(zhuǎn)換等Storm具有以下主要特點：整合性：Storm可方便地與隊列系統(tǒng)和數(shù)據(jù)庫系統(tǒng)進行整合簡易的API：Storm的API在使用上即簡單又方便可擴展性：Storm的并行特性使其可以運行在分布式集群中容錯性：Storm可自動進行故障節(jié)點的重啟、任務(wù)的重新分配可靠的消息處理：Storm保證每個消息都能完整處理支持各種編程語言：Storm支持使用各種編程語言來定義任務(wù)快速部署：Storm可以快速進行部署和使用免費、開源：Storm是一款開源框架，可以免費使用8.4.2Storm的特點Storm可用于許多領(lǐng)域中，如實988.4.3Storm設(shè)計思想Storm主要術(shù)語包括Streams、Spouts、Bolts、Topology和StreamGroupingsStreams：Storm將流數(shù)據(jù)Stream描述成一個無限的Tuple序列，這些Tuple序列會以分布式的方式并行地創(chuàng)建和處理每個tuple是一堆值，每個值有一個名字，并且每個值可以是任何類型Tuple本來應(yīng)該是一個Key-Value的Map，由于各個組件間傳遞的tuple的字段名稱已經(jīng)事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個ValueList（值列表）Field1Field2Field3Field48.4.3Storm設(shè)計思想Storm主要術(shù)語包括Stre998.4.3Storm設(shè)計思想Spout：Storm認為每個Stream都有一個源頭，并把這個源頭抽象為Spout通常Spout會從外部數(shù)據(jù)源（隊列、數(shù)據(jù)庫等）讀取數(shù)據(jù)，然后封裝成Tuple形式，發(fā)送到Stream中。Spout是一個主動的角色，在接口內(nèi)部有個nextTuple函數(shù)，Storm框架會不停的調(diào)用該函數(shù)8.4.3Storm設(shè)計思想Spout：Storm認為每個1008.4.3Storm設(shè)計思想Bolt：Storm將Streams的狀態(tài)轉(zhuǎn)換過程抽象為Bolt。Bolt即可以處理Tuple，也可以將處理后的Tuple作為新的Streams發(fā)送給其他BoltBolt可以執(zhí)行過濾、函數(shù)操作、Join、操作數(shù)據(jù)庫等任何操作Bolt是一個被動的角色，其接口中有一個execute(Tupleinput)方法，在接收到消息之后會調(diào)用此函數(shù)，用戶可以在此方法中執(zhí)行自己的處理邏輯8.4.3Storm設(shè)計思想Bolt：Storm將Stre1018.4.3Storm設(shè)計思想Topology：Storm將Spouts和Bolts組成的網(wǎng)絡(luò)抽象成Topology，它可以被提交到Storm集群執(zhí)行。Topology可視為流轉(zhuǎn)換圖，圖中節(jié)點是一個Spout或Bolt，邊則表示Bolt訂閱了哪個Stream。當(dāng)Spout或者Bolt發(fā)送元組時，它會把元組發(fā)送到每個訂閱了該Stream的Bolt上進行處理Topology里面的每個處理組件（Spout或Bolt）都包含處理邏輯，而組件之間的連接則表示數(shù)據(jù)流動的方向Topology里面的每一個組件都是并行運行的在Topology里面可以指定每個組件的并行度，Storm會在集群里面分配那么多的線程來同時計算在Topology的具體實現(xiàn)上，Storm中的Topology定義僅僅是一些Thrift結(jié)構(gòu)體（二進制高性能的通信中間件），支持各種編程語言進行定義SpoutSpoutBoltBoltBoltBoltBolt8.4.3Storm設(shè)計思想Topology：Storm將1028.4.3Storm設(shè)計思想StreamGroupings：Storm中的StreamGroupings用于告知Topology如何在兩個組件間（如Spout和Bolt之間，或者不同的Bolt之間）進行Tuple的傳送。每一個Spout和Bolt都可以有多個分布式任務(wù)，一個任務(wù)在什么時候、以什么方式發(fā)送Tuple就是由StreamGroupings來決定的8.4.3Storm設(shè)計思想StreamGrouping1038.4.3Storm設(shè)計思想目前，Storm中的StreamGroupings有如下幾種方式：(1)ShuffleGrouping：隨機分組，隨機分發(fā)Stream中的Tuple，保證每個Bolt的Task接收Tuple數(shù)量大致一致(2)FieldsGrouping：按照字段分組，保證相同字段的Tuple分配到同一個Task中(3)AllGrouping：廣播發(fā)送，每一個Task都會收到所有的Tuple(4)GlobalGrouping：全局分組，所有的Tuple都發(fā)送到同一個Task中(5)NonGrouping：不分組，和ShuffleGrouping類似，當(dāng)前Task的執(zhí)行會和它的被訂閱者在同一個線程中執(zhí)行(6)DirectGrouping：直接分組，直接指定由某個Task來執(zhí)行Tuple的處理8.4.3Storm設(shè)計思想目前，Storm中的Strea1048.4.4Storm框架設(shè)計HadoopStorm應(yīng)用名稱JobTopology系統(tǒng)角色JobTrackerNimbusTaskTrackerSupervisor組件接