Spark大數(shù)據(jù)分析 課件 第6章 Spark存儲原理

認(rèn)識SparkStreaming

目錄/Contents01

什么是流式計算01

什么是SparkStreaming01

SparkStreaming工作原理什么是流式計算01什么是流式計算在日常生活中，我們通常會先把數(shù)據(jù)存儲在一張表中，然后再進行加工、分析，這里就涉及到一個時效性的問題。如果我們處理以年、月為單位的級別的數(shù)據(jù)，那么多數(shù)據(jù)的實時性要求并不高；但如果我們處理的是以天、小時，甚至分鐘為單位的數(shù)據(jù)，那么對數(shù)據(jù)的時效性要求就比較高。在第二種場景下，如果我們?nèi)耘f采用傳統(tǒng)的數(shù)據(jù)處理方式，統(tǒng)一收集數(shù)據(jù)，存儲到數(shù)據(jù)庫中，之后在進行分析，就可能無法滿足時效性的要求。1、流式計算誕生背景什么是流式計算流式處理可以用于兩種不同場景：事件流和持續(xù)計算。(1)事件流事件流具能夠持續(xù)產(chǎn)生大量的數(shù)據(jù)，這類數(shù)據(jù)最早出現(xiàn)于傳統(tǒng)的銀行和股票交易領(lǐng)域，也在互聯(lián)網(wǎng)監(jiān)控、無線通信網(wǎng)等領(lǐng)域出現(xiàn)、需要以近實時的方式對更新數(shù)據(jù)流進行復(fù)雜分析如趨勢分析、預(yù)測、監(jiān)控等。簡單來說，事件流采用的是查詢保持靜態(tài)，語句是固定的，數(shù)據(jù)不斷變化的方式。(2)持續(xù)計算比如對于大型網(wǎng)站的流式數(shù)據(jù)：網(wǎng)站的訪問PV/UV、用戶訪問了什么內(nèi)容、搜索了什么內(nèi)容等，實時的數(shù)據(jù)計算和分析可以動態(tài)實時地刷新用戶訪問數(shù)據(jù)，展示網(wǎng)站實時流量的變化情況，分析每天各小時的流量和用戶分布情況；2、流式計算主要應(yīng)用場景什么是SparkStreaming02什么是SparkStreaming1、基本概念SparkStreaming是構(gòu)建在Spark上的實時計算框架，它擴展了Spark處理大規(guī)模流式數(shù)據(jù)的能力。SparkStreaming可結(jié)合批處理和交互查詢，適合一些需要對歷史數(shù)據(jù)和實時數(shù)據(jù)進行結(jié)合分析的應(yīng)用場景。SparkStreaming是Spark的核心組件之一，為Spark提供了可拓展、高吞吐、容錯的流計算能力。如下圖所示，SparkStreaming可整合多種輸入數(shù)據(jù)源，如Kafka、Flume、HDFS，甚至是普通的TCP套接字。經(jīng)處理后的數(shù)據(jù)可存儲至文件系統(tǒng)、數(shù)據(jù)庫，或顯示在儀表盤里。什么是SparkStreamingSparkStreaming是SparkcoreAPI的擴展，支持實時數(shù)據(jù)流的處理，并且具有可擴展，高吞吐量，容錯的特點。SparkStreaming具有如下顯著特點。（1）易用性可以像編寫離線批處理一樣去編寫流式程序，支持java/scala/python編程語言（2）容錯性SparkStreaming在沒有額外代碼和配置的情況下可以恢復(fù)丟失的工作。（3）易整合性流式處理與批處理和交互式查詢相結(jié)合非常方便。什么是SparkStreamingSparkStreaming核心術(shù)語(1)離散流（DStream）(2)批數(shù)據(jù)（batchdata）(3)時間片或批處理時間間隔（batchinterval）(4)窗口長度（windowlength）(5)滑動時間間隔(6)InputDStreamSparkStreaming工作原理03SparkStreaming工作原理SparkStreaming支持從多種數(shù)據(jù)源提取數(shù)據(jù)，如：Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP套接字，并且可以提供一些高級API來表達復(fù)雜的處理算法，如：map、reduce、join和window等。最后，SparkStreaming支持將處理完的數(shù)據(jù)推送到文件系統(tǒng)、數(shù)據(jù)庫或者實時儀表盤中展示。SparkStreaming工作原理整個流式計算根據(jù)業(yè)務(wù)的需求可以對中間的結(jié)果進行緩存或者存儲到外部設(shè)備。感謝大家的聆聽DStream

目錄/Contents01什么是DStream02

DStream轉(zhuǎn)換操作03

DStream輸出04SparkStreaming窗口操作什么是DStream01什么是DStreamSparkStreaming提供了一種高級的抽象，叫做DStream，英文全稱為DiscretizedStream，中文翻譯為“離散流”，它代表了一個持續(xù)不斷的數(shù)據(jù)流。DStream可以通過輸入數(shù)據(jù)源來創(chuàng)建，比如Kafka、Flume和Kinesis；也可以通過對其他DStream應(yīng)用高階函數(shù)來創(chuàng)建，比如map、reduce、join、window。什么是DStreamDStream是spark中繼SparkCore的RDD、SparkSQL的DataFrame和DataSet后又一基礎(chǔ)的數(shù)據(jù)類型，是SparkStreaming特有的數(shù)據(jù)類型。DStream代表了一系列連續(xù)的RDD，DStream中每個RDD包含特定時間間隔的數(shù)據(jù)，存儲方式為HashMap<Time,RDD>。其中，Time為時間序列，而RDD我們都很熟悉，它是SparkCore的基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)。DStream轉(zhuǎn)換操作02DStream轉(zhuǎn)換操作方法名稱描述map(func)通過將源DStream的每個元素傳遞給函數(shù)func來返回新的DStream

。flatMap(func)與map相似，不同的是每個輸入元素可以被映射出0或者更多的輸出元素。filter(func)通過僅選擇func返回true

的源DStream的記錄來返回新的DStream

。repartition(numPartitions)通過創(chuàng)建更多或更少的分區(qū)來更改此DStream中的并行度。union(otherStream)返回一個新的DStream，其中包含源DStream和otherDStream中的元素的并集。count()通過計算源DStream的每個RDD中的元素數(shù)，返回一個新的單元素RDDDStream。

reduce(func)通過使用函數(shù)func（帶有兩個參數(shù)并返回一個）來聚合源DStream的每個RDD中的元素，從而返回一個單元素RDD的新DStream

。該函數(shù)應(yīng)具有關(guān)聯(lián)性和可交換性，以便可以并行計算。

countByValue()在類型為K的元素的DStream上調(diào)用時，返回一個新的（K，Long）對的DStream，其中每個鍵的值是其在源DStream的每個RDD中的頻率。reduceByKey(func,[numTasks])在（K，V）對的DStream上調(diào)用時，返回一個新的（K，V）對的DStream，其中使用給定的reduce函數(shù)聚合每個鍵的值。注意：默認(rèn)情況下，這使用Spark的默認(rèn)并行任務(wù)數(shù)（本地模式為2，而在集群模式下，此數(shù)量由config屬性確定spark.default.parallelism）進行分組。您可以傳遞一個可選numTasks參數(shù)來設(shè)置不同數(shù)量的任務(wù)。

join(otherStream,[numTasks])當(dāng)在（K，V）和（K，W）對的兩個DStream上調(diào)用時，返回一個新的（K，（V，W））對的DStream，其中每個鍵都有所有元素對。

cogroup(otherStream,[numTasks])在（K，V）和（K，W）對的DStream上調(diào)用時，返回一個新的（K，Seq[V]，Seq[W]）元組的DStream。

transform(func)通過對源DStream的每個RDD應(yīng)用RDD-to-RDD函數(shù)來返回新的DStream。這可用于在DStream上執(zhí)行任意RDD操作。

updateStateByKey(func)返回一個新的“狀態(tài)”DStream，在該DStream中，通過在鍵的先前狀態(tài)和鍵的新值上應(yīng)用給定函數(shù)來更新每個鍵的狀態(tài)。這可用于維護每個鍵的任意狀態(tài)數(shù)據(jù)。什么是DStream1.transform(func)在Spark-Streaming官方文檔中提到，DStream的transform操作極大的豐富了DStream上能夠進行的操作內(nèi)容。使用transform操作后，除了可以使用DStream提供的一些轉(zhuǎn)換方法之外，還能夠直接調(diào)用任意的調(diào)用RDD上的操作函數(shù)。下面演示如何使用transform將一行語句切分成多個單詞。2.updateStateByKey(func)updateStateByKey會統(tǒng)計全局的key的狀態(tài)，不管又沒有數(shù)據(jù)輸入，它會在每一個批次間隔返回之前的key的狀態(tài)。updateStateByKey會對已存在的key進行state的狀態(tài)更新，同時還會對每個新出現(xiàn)的key執(zhí)行相同的更新函數(shù)操作。DStream輸出03DStream輸出OutputOperations可以將DStream的數(shù)據(jù)輸出到外部的數(shù)據(jù)庫或文件系統(tǒng)，當(dāng)某個OutputOperations原語被調(diào)用時（與RDD的Action相同），streaming程序才會開始真正的計算過程。也就是說在SparkStreaming中，DStream的輸出操作是真正觸發(fā)DStream上所有轉(zhuǎn)換操作進行計算(類似于RDD中的Action算子操作)的操作，然后經(jīng)過輸出操作，DStream中的數(shù)據(jù)才能與外部進行交互，如將數(shù)據(jù)寫入到分布式文件系統(tǒng)、數(shù)據(jù)庫以及其他應(yīng)用中。DStream輸出方法描述

print()在運行流應(yīng)用程序的驅(qū)動程序節(jié)點上，打印DStream中每批數(shù)據(jù)的前十個元素。這對于開發(fā)和調(diào)試很有用。

saveAsTextFiles(prefix,[suffix])將此DStream的內(nèi)容另存為文本文件?；诋a(chǎn)生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

saveAsObjectFiles(prefix,[suffix])將此DStream的內(nèi)容保存為SequenceFiles序列化Java對象的內(nèi)容。基于產(chǎn)生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

saveAsHadoopFiles(prefix,[suffix])將此DStream的內(nèi)容另存為Hadoop文件?；诋a(chǎn)生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

foreachRDD(func)最通用的輸出運算符，將函數(shù)func應(yīng)用于從流生成的每個RDD。此功能應(yīng)將每個RDD中的數(shù)據(jù)推送到外部系統(tǒng)，例如將RDD保存到文件或通過網(wǎng)絡(luò)將其寫入數(shù)據(jù)庫。請注意，函數(shù)func在運行流應(yīng)用程序的驅(qū)動程序進程中執(zhí)行，并且通常在其中具有RDD操作，這將強制計算流RDD。SparkStreaming窗口操作04SparkStreaming窗口操作方法描述

window(windowLength,

slideInterval)

返回基于源DStream的窗口批處理計算的新DStream。countByWindow(windowLength,

slideInterval)返回基于滑動窗口的DStream中的元素數(shù)。

reduceByWindow(func,

windowLength,

slideInterval)返回一個新的單元素流，該流是通過使用func在滑動間隔內(nèi)聚合流中的元素而創(chuàng)建的。該函數(shù)應(yīng)該是關(guān)聯(lián)的和可交換的，以便可以并行正確地計算它。

reduceByKeyAndWindow(func,

windowLength,

slideInterval,[numTasks])在（K，V）對的DStream上調(diào)用時，返回新的（K，V）對的DStream，其中使用給定的reduce函數(shù)func在滑動窗口中的批處理上聚合每個鍵的值。注意：默認(rèn)情況下，這使用Spark的默認(rèn)并行任務(wù)數(shù)（本地模式為2，而在集群模式下，此數(shù)量由config屬性確定spark.default.parallelism）進行分組。您可以傳遞一個可選numTasks參數(shù)來設(shè)置不同數(shù)量的任務(wù)。

reduceByKeyAndWindow(func,

invFunc,

windowLength,

slideInterval,[numTasks])一種更有效的版本，reduceByKeyAndWindow()其中，使用前一個窗口的減少值遞增地計算每個窗口的減少值。這是通過減少進入滑動窗口的新數(shù)據(jù)并“逆向減少”離開窗口的舊數(shù)據(jù)來完成的。一個示例是在窗口滑動時“增加”和“減少”鍵的計數(shù)。但是，它僅適用于“可逆歸約函數(shù)”，即具有對應(yīng)的“逆歸約”函數(shù)（作為參數(shù)invFunc）的歸約函數(shù)。像in中一樣reduceByKeyAndWindow，reduce任務(wù)的數(shù)量可以通過可選參數(shù)配置。請注意，必須啟用檢查點才能使用此操作。

countByValueAndWindow(windowLength,

slideInterval,[numTasks])在（K，V）對的DStream上調(diào)用時，返回新的（K，Long）對的DStream，其中每個鍵的值是其在滑動窗口內(nèi)的頻率。像in中一樣

reduceByKeyAndWindow，reduce任務(wù)的數(shù)量可以通過可選參數(shù)配置。感謝大家的聆聽SparkStreaming應(yīng)用案例目錄/Contents01SparkStreaming實現(xiàn)單詞統(tǒng)計02SparkStreaming熱搜統(tǒng)計03自定義輸出04SparkStreaming窗口計算SparkStreaming實現(xiàn)單詞統(tǒng)計01SparkStreaming實現(xiàn)單詞統(tǒng)計（1）實驗內(nèi)容本實驗主要介紹在IDEA中開發(fā)SparkStreaming案例-WordCount，從TCPSocket數(shù)據(jù)源實時消費數(shù)據(jù)，對每批次Batch數(shù)據(jù)進行詞頻統(tǒng)計。（2）實驗?zāi)康?、掌握Spark窗口計算（3）實驗步驟步驟1準(zhǔn)備工作步驟2SparkStreaming-WordCount代碼實現(xiàn)（4）步驟1準(zhǔn)備工作在任一主機上安裝nc命令：yuminstall-yncnc是netcat的簡稱，原本是用來設(shè)置路由器，我們可以利用它向某個端口發(fā)送數(shù)據(jù)。（5）步驟2SparkStreaming-WordCount代碼實現(xiàn)在spark_wc項目下新建名為WordCount01的scala類，使用SparkStreaming接收來自某一主機9999端口的數(shù)據(jù)并做WordCount。SparkStreaming熱搜統(tǒng)計02SparkStreaming熱搜統(tǒng)計（1）實驗內(nèi)容本實驗主要介紹在IDEA中開發(fā)SparkStreaming案例-WordCount，從TCPSocket數(shù)據(jù)源實時消費數(shù)據(jù)，模擬百度熱搜排行榜。（2）實驗?zāi)康?、熟悉DStream的transform方法的使用（3）實驗步驟步驟1代碼實現(xiàn)（4）步驟1代碼實現(xiàn)在spark_wc項目下新建名為WordCount05的scala類，使用