Spark大數(shù)據(jù)分析 課件 6.2 DStream

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）進(jìn)行分組。您可以傳遞一個可選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)。這可用于維護(hù)每個鍵的任意狀態(tài)數(shù)據(jù)。什么是DStream1.transform(func)在Spark-Streaming官方文檔中提到，DStream的transform操作極大的豐富了DStream上能夠進(jìn)行的操作內(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進(jìn)行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)換操作進(jìn)行計算(類似于RDD中的Action算子操作)的操作，然后經(jīng)過輸出操作，DStream中的數(shù)據(jù)才能與外部進(jìn)行交互，如將數(shù)據(jù)寫入到分布式文件系統(tǒng)、數(shù)據(jù)庫以及其他應(yīng)用中。DStream輸出方法描述

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

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

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

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

foreachRDD(func)最通用的輸出運(yùn)算符，將函數(shù)func應(yīng)用于從流生成的每個RDD。此功能應(yīng)將每個RDD中的數(shù)據(jù)推送到外部系統(tǒng)，例如將RDD保存到文件或通過網(wǎng)絡(luò)將其寫入數(shù)據(jù)庫。請注意，函數(shù)func在運(yùn)行流應(yīng)用程序的驅(qū)動程序進(jìn)程中執(zhí)行，并且通常在其中具有RDD操作，這將強(qiáng)制計算流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）進(jìn)行分組。您可以傳遞一個可選numTasks參數(shù)來設(shè)置不同數(shù)量的任務(wù)。

reduceByKeyAndWindow(func,

invFunc,

windowLength,

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