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

DStream

目錄/Contents01什么是DStream02

DStream轉(zhuǎn)換操作03

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

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

的源DStream的記錄來(lái)返回新的DStream

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

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

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

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

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

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

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

updateStateByKey(func)返回一個(gè)新的“狀態(tài)”DStream，在該DStream中，通過(guò)在鍵的先前狀態(tài)和鍵的新值上應(yīng)用給定函數(shù)來(lái)更新每個(gè)鍵的狀態(tài)。這可用于維護(hù)每個(gè)鍵的任意狀態(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將一行語(yǔ)句切分成多個(gè)單詞。2.updateStateByKey(func)updateStateByKey會(huì)統(tǒng)計(jì)全局的key的狀態(tài)，不管又沒(méi)有數(shù)據(jù)輸入，它會(huì)在每一個(gè)批次間隔返回之前的key的狀態(tài)。updateStateByKey會(huì)對(duì)已存在的key進(jìn)行state的狀態(tài)更新，同時(shí)還會(huì)對(duì)每個(gè)新出現(xiàn)的key執(zhí)行相同的更新函數(shù)操作。DStream輸出03DStream輸出OutputOperations可以將DStream的數(shù)據(jù)輸出到外部的數(shù)據(jù)庫(kù)或文件系統(tǒng)，當(dāng)某個(gè)OutputOperations原語(yǔ)被調(diào)用時(shí)（與RDD的Action相同），streaming程序才會(huì)開(kāi)始真正的計(jì)算過(guò)程。也就是說(shuō)在SparkStreaming中，DStream的輸出操作是真正觸發(fā)DStream上所有轉(zhuǎn)換操作進(jìn)行計(jì)算(類似于RDD中的Action算子操作)的操作，然后經(jīng)過(guò)輸出操作，DStream中的數(shù)據(jù)才能與外部進(jìn)行交互，如將數(shù)據(jù)寫(xiě)入到分布式文件系統(tǒng)、數(shù)據(jù)庫(kù)以及其他應(yīng)用中。DStream輸出方法描述

print()在運(yùn)行流應(yīng)用程序的驅(qū)動(dòng)程序節(jié)點(diǎn)上，打印DStream中每批數(shù)據(jù)的前十個(gè)元素。這對(duì)于開(kāi)發(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對(duì)象的內(nèi)容?；诋a(chǎn)生在每批間隔的文件名的前綴和后綴：“prefix-TIME_IN_MS[.suffix]”。

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

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

window(windowLength,

slideInterval)

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

slideInterval)返回基于滑動(dòng)窗口的DStream中的元素?cái)?shù)。

reduceByWindow(func,

windowLength,

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

reduceByKeyAndWindow(func,

windowLength,

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

reduceByKeyAndWindow(func,

invFunc,

windowLength,

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