《Spark應(yīng)用開發(fā)技術(shù)》課件-02-掌握Dstram編程模型

Spark應(yīng)用開發(fā)技術(shù)本章主要講述SparkStreaming實(shí)時(shí)計(jì)算框架、Dstream編程模型。通過學(xué)習(xí)本節(jié)將能夠?qū)W習(xí)SparkStreaming實(shí)時(shí)計(jì)算框架、Dstream編程模型。通過本節(jié)學(xué)習(xí)可以：理解SparkStreaming的基本概念及運(yùn)行原理掌握Dstream的轉(zhuǎn)換操作掌握Dstream的窗口操作掌握Dstream的輸出操作初探SparkStreamingSparkStreaming簡(jiǎn)介Spark

Streaming運(yùn)行原理初步使用SparkStreaming掌握Dstream編程模型Dstream簡(jiǎn)介Dstream轉(zhuǎn)換操作Dstream窗口操作Dstream輸出操作3.

Spark

Streaming實(shí)時(shí)更新熱門博文DStream簡(jiǎn)介SparkStreaming提供了一個(gè)高層次的抽象叫做離散流（discretizedstream）或者DStream，代表了持續(xù)性的數(shù)據(jù)流DStream可以通過外部數(shù)據(jù)源（Kafka，F(xiàn)lume，Twitter等）來(lái)獲取，也可以通過現(xiàn)有DStream的高級(jí)操作（Transformation操作）獲得在內(nèi)部，DStream代表著一系列的持續(xù)的RDDsDStream中的每個(gè)RDD都是一小段時(shí)間（interval）分割開來(lái)的數(shù)據(jù)集掌握Dstream編程模型DStream簡(jiǎn)介對(duì)DStream的任何操作都會(huì)轉(zhuǎn)化成對(duì)底層RDDs的操作例如，對(duì)linesDStream做flatMap操作，實(shí)際上就是對(duì)它內(nèi)部的所有RDD做flatMap操作。即對(duì)DStream的操作可以通過RDD的Transformation生成新的DStream。掌握Dstream編程模型DStream轉(zhuǎn)換操作掌握Dstream編程模型DStream轉(zhuǎn)換操作掌握Dstream編程模型DStream轉(zhuǎn)換操作以上列出的這些操作，大部分（如map，flatMap，filter等）與RDD的轉(zhuǎn)換操作類似transform(func)方法值得深入地探討transform操作極大地豐富了DStream上能夠進(jìn)行的操作內(nèi)容。使用transform操作后，除了可以使用DStream提供的一些轉(zhuǎn)換方法之外，還能夠直接調(diào)用任意RDD上的操作方法。掌握Dstream編程模型DStream轉(zhuǎn)換操作使用transform將一行語(yǔ)句分割成單詞掌握Dstream編程模型DStream轉(zhuǎn)換操作在slave18888端口上輸入“IamlearningSparkStreamingnow”語(yǔ)句運(yùn)行結(jié)果如圖所示，該語(yǔ)句在5s內(nèi)被分割成單詞掌握Dstream編程模型DStream窗口操作窗口函數(shù)，就是在DStream流上，以一個(gè)可配置的長(zhǎng)度為窗口，以一個(gè)可配置的速率向前移動(dòng)窗口，根據(jù)窗口函數(shù)的具體內(nèi)容，對(duì)窗口內(nèi)的數(shù)據(jù)執(zhí)行計(jì)算操作，每次掉落在窗口內(nèi)的RDD的數(shù)據(jù)會(huì)被聚合起來(lái)執(zhí)行計(jì)算操作，然后生成的RDD會(huì)作為WindowDStream的一個(gè)RDD。下圖表述的是滑動(dòng)窗口長(zhǎng)度為3秒，這三秒內(nèi)的3個(gè)RDD會(huì)被聚合起來(lái)進(jìn)行處理，然后過了兩秒鐘，又會(huì)對(duì)最近三秒內(nèi)的數(shù)據(jù)執(zhí)行滑動(dòng)窗口計(jì)算。所以每個(gè)滑動(dòng)窗口操作，都必須指定兩個(gè)參數(shù)，窗口長(zhǎng)度以及滑動(dòng)間隔，而且這兩個(gè)參數(shù)值都必須是batch（批處理時(shí)間）間隔的整數(shù)倍。掌握Dstream編程模型DStream窗口操作常用的窗口轉(zhuǎn)換操作方法如表所示這些操作都需要兩個(gè)參數(shù)，windowLength（窗口長(zhǎng)度）和slideInterval（時(shí)間間隔）。掌握Dstream編程模型DStream窗口操作掌握Dstream編程模型DStream窗口操作以window(windowLength,slideInterval)為例設(shè)置窗口長(zhǎng)度為3s，滑動(dòng)時(shí)間間隔為1s，截取源DStream中的元素形成新的DStream掌握Dstream編程模型DStream窗口操作掌握Dstream編程模型在slave1上啟動(dòng)監(jiān)聽，基本上每秒輸入一個(gè)字母，然后取出當(dāng)前時(shí)刻3秒這個(gè)長(zhǎng)度中的所有元素，打印出來(lái)如圖所示，從圖中可以看到，到第4秒時(shí)已經(jīng)不到a了，再下一秒就看不到b了，這說明此時(shí)a和b已經(jīng)不在當(dāng)前窗口中。DStream窗口操作reduceByKeyAndWindow(func,windowLength,slideInterval,[numTasks])操作類似reduceByKey操作，只不過兩者操作的數(shù)據(jù)源不同，reduceByKeyAndWindow的數(shù)據(jù)源是基于該DStream的窗口長(zhǎng)度中的所有數(shù)據(jù)。將當(dāng)前長(zhǎng)度為3的時(shí)間窗口中的所有數(shù)據(jù)元素根據(jù)key進(jìn)行合并，統(tǒng)計(jì)當(dāng)前3秒中內(nèi)不同單詞出現(xiàn)的次數(shù)。掌握Dstream編程模型DStream窗口操作掌握Dstream編程模型DStream窗口操作掌握Dstream編程模型運(yùn)行結(jié)果如圖所示從圖中可以看出，到了第4s，最前面的兩個(gè)a已經(jīng)不在當(dāng)前窗口中，所以沒有打印出a的計(jì)數(shù)。DStream輸出操作掌握Dstream編程模型DStream輸出操作saveAsTextFiles、saveAsObjectFiles和saveAsHadoopFiles操作可以將DStream中的內(nèi)容保存為文本文件每個(gè)batch的數(shù)據(jù)單獨(dú)保存為一個(gè)文夾，其中prefix為文件夾名前綴，文件夾名前綴參數(shù)必須傳入，[suffix]為文件夾名后綴，文件夾名后綴參數(shù)可選，最終文件夾名稱的完整形式為prefix-TIME_IN_MS[.suffix]。如果前綴中包含文件完整路徑，則該text文件夾會(huì)建在指定路徑下。saveAsTextFiles以文本的形式保存DStream中的內(nèi)容，可以保存在任何文件系統(tǒng)saveAsObjectFiles是以序列化的格式保存saveAsHadoopFiles是以文本的形式保存在HDFS上。掌握Dstream編程模型DStream輸出操作將nc窗口中輸出的內(nèi)容保存在HDFS的/user/root/saveAsTextFiles文件夾下，設(shè)置每秒生成一個(gè)文件夾掌握Dstream編程模型DStream輸出操作保存結(jié)果如圖所示掌握Dstream編程模型DStream輸出操作foreachRDD是DStream提供的一個(gè)功能強(qiáng)大的方法，它可以將數(shù)據(jù)發(fā)送到外部系統(tǒng)，在使用foreachRDD的過程中需避免以下錯(cuò)誤通常將數(shù)據(jù)寫入到外部系統(tǒng)需要?jiǎng)?chuàng)建一個(gè)連接對(duì)象（如TCP連接到遠(yuǎn)程服務(wù)器），并用它來(lái)發(fā)送數(shù)據(jù)到遠(yuǎn)程系統(tǒng)。在創(chuàng)建連接對(duì)象時(shí)應(yīng)避免在Sparkdriver端創(chuàng)建連接對(duì)象，代碼如下所示，這種做法需要連接對(duì)象進(jìn)行序列化并從Driver端發(fā)送到Worker上，但是連接對(duì)象很少在不同機(jī)器間進(jìn)行這種操作掌握Dstream編程模型DStream輸出操作foreachRDD是DStream提供的一個(gè)功能強(qiáng)大的方法，它可以將數(shù)據(jù)發(fā)送到外部系統(tǒng)，在使用foreachRDD的過程中需避免以下錯(cuò)誤針對(duì)以上所說的錯(cuò)誤，正確的解決方法是在Worker上創(chuàng)建連接對(duì)象，如代碼所示，但是這種做法又會(huì)引發(fā)另外一種錯(cuò)誤，即為每一個(gè)記錄創(chuàng)建一個(gè)連接對(duì)象。通常，創(chuàng)建一個(gè)連接對(duì)象會(huì)有時(shí)間和資源的開銷，因此，為每個(gè)記錄創(chuàng)建和銷毀連接對(duì)象會(huì)導(dǎo)致非常高的開銷，減少系統(tǒng)的整體吞吐量。掌握Dstream編程模型DStream輸出操作foreachRDD的正確用法：使用rdd.foreachPartition方法創(chuàng)建一個(gè)單獨(dú)的連接對(duì)象，然后使用該連接對(duì)象輸出所有RDD分區(qū)中的數(shù)據(jù)到外部系統(tǒng)。這不僅可以緩解創(chuàng)建多條記錄連接的開銷，還可以通過在多個(gè)RDDs/batches上重用連接對(duì)象進(jìn)行優(yōu)化。掌握Dstream編程模型DStream輸出操作以網(wǎng)站熱詞排名為例，介紹如何正確使用foreachPartition將處理結(jié)果寫到MySQL數(shù)據(jù)庫(kù)中。首先在MySQL數(shù)據(jù)庫(kù)中創(chuàng)建數(shù)據(jù)庫(kù)和表用以接收處理后的數(shù)據(jù)新建的表searchKeyWord表有三個(gè)字段，分別為insert_date（插入數(shù)據(jù)的日期），keyword（熱詞），search_count（在設(shè)置的時(shí)間內(nèi)出現(xiàn)的次數(shù)）。掌握Dstream編程模型DStream輸出操作在IntelliJIDEA中編寫Spark代碼，設(shè)置窗口長(zhǎng)度為60秒，窗口滑動(dòng)時(shí)間間隔為10秒，計(jì)算10秒內(nèi)每個(gè)單詞出現(xiàn)的次數(shù)，然后根據(jù)出現(xiàn)的次數(shù)對(duì)單詞進(jìn)行排序雖然DStream沒有提供sort的方法，但是可以使用transform函數(shù)，用RDD的sortByKey實(shí)現(xiàn)。接著需要使用foreachPartition創(chuàng)建MySQL數(shù)據(jù)庫(kù)連接對(duì)象然后使用該連接對(duì)象輸出數(shù)據(jù)到searchKeyWord表中掌握Dstream編程模型DStream輸出操作