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可以通過輸入數(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中每個(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)通過將源DStream的每個(gè)元素傳遞給函數(shù)func來返回新的DStream

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

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

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

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

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

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

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

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

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

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

print()在運(yùn)行流應(yīng)用程序的驅(qū)動(dòng)程序節(jié)點(diǎn)上，打印DStream中每批數(shù)據(jù)的前十個(gè)元素。這對于開發(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)容?；诋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)用于從流生成的每個(gè)RDD。此功能應(yīng)將每個(gè)RDD中的數(shù)據(jù)推送到外部系統(tǒng)，例如將RDD保存到文件或通過網(wǎng)絡(luò)將其寫入數(shù)據(jù)庫。請注意，函數(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è)新的單元素流，該流是通過使用func在滑動(dòng)間隔內(nèi)聚合流中的元素而創(chuàng)建的。該函數(shù)應(yīng)該是關(guān)聯(lián)的和可交換的，以便可以并行正確地計(jì)算它。

reduceByKeyAndWindow(func,

windowLength,

slideInterval,[numTasks])在（K，V）對的DStream上調(diào)用時(shí)，返回新的（K，V）對的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ù)來設(shè)置不同數(shù)量的任務(wù)。

reduceByKeyAndWindow(func,

invFunc,

windowLength,

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