Spark大數據分析 課件 第6章 Spark存儲原理

認識SparkStreaming

目錄/Contents01

什么是流式計算01

什么是SparkStreaming01

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

目錄/Contents01什么是DStream02

DStream轉換操作03

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

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

。該函數應具有關聯(lián)性和可交換性，以便可以并行計算。

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

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

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

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

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

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

saveAsTextFiles(prefix,[suffix])將此DStream的內容另存為文本文件。基于產生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

saveAsObjectFiles(prefix,[suffix])將此DStream的內容保存為SequenceFiles序列化Java對象的內容?；诋a生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

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

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

window(windowLength,

slideInterval)

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

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

reduceByWindow(func,

windowLength,

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

reduceByKeyAndWindow(func,

windowLength,

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

reduceByKeyAndWindow(func,

invFunc,

windowLength,

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

countByValueAndWindow(windowLength,

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

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