SparkStreaming

——實(shí)時(shí)計(jì)算框架書(shū)籍是人類進(jìn)步的階梯，數(shù)字時(shí)代的來(lái)臨，也催生出“書(shū)”的新形式，即電子書(shū)。同時(shí)，眾多售書(shū)的電商平臺(tái)也應(yīng)運(yùn)而生。電商平臺(tái)想要在激烈的競(jìng)爭(zhēng)中脫穎而出，需要更著重于改善用戶體驗(yàn)，并增加用戶的黏性，把更多更好的書(shū)推薦給讀者，擴(kuò)大他們的知識(shí)視野。用戶無(wú)法找到適宜的書(shū)籍時(shí)往往會(huì)相信大眾的選擇，選擇購(gòu)買(mǎi)熱度較高的書(shū)籍?；谶@種情況，電商平臺(tái)可以根據(jù)現(xiàn)有書(shū)籍的評(píng)分、銷量、用戶的評(píng)分次數(shù)等信息構(gòu)建書(shū)籍熱度，將一些熱度較高的書(shū)推薦給用戶，進(jìn)而改善用戶體驗(yàn)，增加用戶黏性，激發(fā)用戶的購(gòu)買(mǎi)欲。任務(wù)背景書(shū)籍熱度的計(jì)算可以根據(jù)下式進(jìn)行，其中，u表示用戶的平均評(píng)分，x表示用戶的評(píng)分次數(shù)，y表示書(shū)籍的平均評(píng)分，z表示書(shū)籍被評(píng)分的次數(shù)。目前已采集了某電商網(wǎng)站上用戶對(duì)書(shū)籍的評(píng)分?jǐn)?shù)據(jù)文件BookRating.txt，數(shù)據(jù)字段說(shuō)明如下表。其中Rating字段中評(píng)分范圍為1～5分。任務(wù)背景字段名稱說(shuō)明UserID用戶IDBookID書(shū)籍IDRating用戶對(duì)書(shū)籍的評(píng)分實(shí)時(shí)計(jì)算書(shū)籍熱度后，可以將熱度最高的10本圖書(shū)的評(píng)分?jǐn)?shù)據(jù)保存在Hive數(shù)據(jù)庫(kù)中，因此需要在Hive數(shù)據(jù)庫(kù)中設(shè)計(jì)一個(gè)表，用于保存熱度最高的10本圖書(shū)的評(píng)分?jǐn)?shù)據(jù)。Spark會(huì)將DataFrame寫(xiě)入Hive并根據(jù)DataFrame自動(dòng)創(chuàng)建表。為模擬實(shí)時(shí)數(shù)據(jù)的流式計(jì)算，本章將使用SparkStreaming框架實(shí)現(xiàn)書(shū)籍評(píng)分實(shí)時(shí)計(jì)算分析。本章任務(wù)如下。首先介紹SparkStreaming基本概念及運(yùn)行原理；再詳細(xì)介紹SparkStreaming框架的DStream編程模型及其基礎(chǔ)操作；最后結(jié)合書(shū)籍評(píng)分?jǐn)?shù)據(jù)實(shí)例，使用SparkStreaming框架實(shí)現(xiàn)書(shū)籍熱度的實(shí)時(shí)計(jì)算。任務(wù)背景1掌握DStream基礎(chǔ)操作目錄初識(shí)SparkStreaming2實(shí)現(xiàn)書(shū)籍熱度實(shí)時(shí)計(jì)算3使用SparkStreaming實(shí)現(xiàn)書(shū)籍熱度實(shí)時(shí)計(jì)算，首先需要對(duì)SparkStreaming基本概念及運(yùn)行原理有大致的了解。本節(jié)的任務(wù)如下，了解SparkStreaming基本概念及運(yùn)行原理；學(xué)習(xí)SparkStreaming程序的簡(jiǎn)單編寫(xiě)及運(yùn)行。任務(wù)描述SparkStreaming是Spark的子框架，是Spark生態(tài)圈中用于處理流式數(shù)據(jù)的分布式流式處理框架，具有可伸縮、高吞吐量、容錯(cuò)能力強(qiáng)等特點(diǎn)。同時(shí)，SparkStreaming能夠和SparkSQL、SparkMLlib、SparkGraphX進(jìn)行無(wú)縫集成，可以從Kafka、Flume、HDFS、Kinesis等數(shù)據(jù)源中獲取數(shù)據(jù)，而且不僅可以通過(guò)調(diào)用map()、reduce()、join()等方法處理數(shù)據(jù)，也可以使用機(jī)器學(xué)習(xí)算法、圖算法處理數(shù)據(jù)。如右圖，經(jīng)SparkStreaming處理后的最終結(jié)果可以保存在文件系統(tǒng)（如HDFS）、數(shù)據(jù)庫(kù)（如MySQL）中或使用儀表面板進(jìn)行實(shí)時(shí)展示。了解SparkStreaming基本概念SparkStreaming的運(yùn)行原理圖了解SparkStreaming運(yùn)行原理使用SparkStreaming一般需要進(jìn)行如下的操作。創(chuàng)建StreamingContext對(duì)象。創(chuàng)建InputDStream。操作Dstream。啟動(dòng)SparkStreaming。初步使用SparkStreaming單詞實(shí)時(shí)計(jì)數(shù)從一臺(tái)服務(wù)器的8888端口上接受一行或者多行文本內(nèi)容，并對(duì)接收到的內(nèi)容以空格分割計(jì)算每個(gè)單詞出現(xiàn)的次數(shù)。初步使用SparkStreaming單詞實(shí)時(shí)計(jì)數(shù)初步使用SparkStreaming使用ssc.textFileStream()方法監(jiān)聽(tīng)HDFS上的目錄/user/root/sparkStreaming/temp，一旦有新文件加入到/user/root/sparkStreaming/temp目錄下，SparkStreaming計(jì)算出該時(shí)間內(nèi)的單詞統(tǒng)計(jì)數(shù)。初步使用SparkStreamingSparkStreaming監(jiān)聽(tīng)HDFS目錄示例運(yùn)行結(jié)果初步使用SparkStreaming1掌握DStream基礎(chǔ)操作目錄初識(shí)SparkStreaming2實(shí)現(xiàn)書(shū)籍熱度實(shí)時(shí)計(jì)算3DStream是SparkStreaming中一個(gè)非常重要的概念。SparkStreaming讀取數(shù)據(jù)時(shí)會(huì)得到DStream編程模型，且DStream提供了一系列操作方法。本節(jié)的任務(wù)如下。了解DStream的基本概念。學(xué)習(xí)DStream的轉(zhuǎn)換操作、窗口操作以及輸出操作。任務(wù)描述DStream是SparkStreaming對(duì)內(nèi)部實(shí)時(shí)數(shù)據(jù)流的抽象描述，可將DStream理解為持續(xù)性的數(shù)據(jù)流。可以通過(guò)外部數(shù)據(jù)源獲取DStream；也可以通過(guò)DStream現(xiàn)有的高級(jí)操作（如轉(zhuǎn)換操作）獲得DStream。DStream代表著一系列的持續(xù)的RDD，DStream中的每個(gè)RDD都是按一小段時(shí)間分割開(kāi)的RDD，如下圖。了解DStream編程模型對(duì)DStream的任何操作都會(huì)轉(zhuǎn)化成對(duì)底層RDDs的操作。以單詞計(jì)數(shù)為例，獲取文本數(shù)據(jù)形成文本的輸入數(shù)據(jù)流linesDStream，使用flatMap()方法進(jìn)行扁平化操作并進(jìn)行分割，得到每一個(gè)單詞，形成單詞的文本數(shù)據(jù)流wordsDStream。對(duì)DStream進(jìn)行操作的方法根據(jù)操作的類型可以分成3類，即轉(zhuǎn)換操作、窗口操作和輸出操作。了解DStream編程模型使用DStream轉(zhuǎn)換操作DStream轉(zhuǎn)換操作常用的方法及說(shuō)明方法描述map(func)對(duì)源DStream的每個(gè)元素應(yīng)用func函數(shù)并返回一個(gè)新的DStreamflatMap(func)類似map操作，不同的是每個(gè)元素可以被映射成0個(gè)或者多個(gè)輸出元素filter(func)對(duì)源DStream中的每一個(gè)元素應(yīng)用func函數(shù)進(jìn)行計(jì)算，如果func函數(shù)返回結(jié)果為true，則保留該元素，否則丟棄該元素，返回一個(gè)新的DStreamunion(otherStream)合并兩個(gè)DStream，生成一個(gè)包含兩個(gè)DStream中所有元素的新的DStreamcount()統(tǒng)計(jì)DStream中每個(gè)RDD包含的元素的個(gè)數(shù)，得到一個(gè)只有一個(gè)元素的RDD構(gòu)成的DStreamreduce(func)對(duì)源DStream中的每個(gè)元素應(yīng)用func函數(shù)進(jìn)行聚合操作，返回一個(gè)內(nèi)部所包含的RDD只有一個(gè)元素的新DStream使用DStream轉(zhuǎn)換操作DStream轉(zhuǎn)換操作常用的方法及說(shuō)明方法描述countByKey()計(jì)算DStream中每個(gè)RDD內(nèi)的元素出現(xiàn)的頻次，并返回新的DStream[(K,Long)]，其中K是RDD中元素的類型，Long是元素出現(xiàn)的頻次reduceByKey(func,[numTasks])以一個(gè)鍵值RDD為目標(biāo)，K為鍵，V為值。當(dāng)一個(gè)(K，V)鍵值對(duì)的DStream被調(diào)用時(shí)，返回(K,V)鍵值對(duì)的新DStream，其中每個(gè)鍵的值都使用聚合函數(shù)func匯總。配置numTasks可以設(shè)置不同的并行任務(wù)數(shù)join(otherStream,[numTasks])當(dāng)調(diào)用的是(K,V1)和(K,V2)鍵值對(duì)的兩個(gè)DStream時(shí)，返回元素為(K,(V1,V2))鍵值對(duì)的一個(gè)新DStreamcogroup(otherStream,[numTasks])當(dāng)被調(diào)用的兩個(gè)DStream分別含有(K,V1)和(K,V2)鍵值對(duì)時(shí)，返回一個(gè)元素為(K,Seq[V1],Seq[V2])的新的DStreamtransform(func)通過(guò)對(duì)源DStream的每個(gè)RDD應(yīng)用func函數(shù)返回一個(gè)新的DStream，用于在DStream上進(jìn)行RDD的任意操作大部分操作（如map，flatMap，filter等）與RDD的轉(zhuǎn)換操作類似。transform操作極大地豐富了DStream上能夠進(jìn)行的操作內(nèi)容。使用transform操作后，除了可以使用DStream提供的一些轉(zhuǎn)換方法之外，還能夠直接調(diào)用任意RDD上的操作方法。使用DStream轉(zhuǎn)換操作使用transform將一行語(yǔ)句分割成單詞使用DStream轉(zhuǎn)換操作在slave18888端口上輸入“IamlearningSparkStreamingnow”語(yǔ)句。運(yùn)行結(jié)果如下圖，該語(yǔ)句在5s內(nèi)被分割成單詞。使用DStream轉(zhuǎn)換操作窗口函數(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)行處理，然后過(guò)了兩秒鐘，又會(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窗口操作常用的窗口轉(zhuǎn)換操作方法如下表。這些操作都需要兩個(gè)參數(shù)，windowLength（窗口長(zhǎng)度）和slideInterval（時(shí)間間隔）。使用DStream窗口操作方法描述window(windowLength,slideInterval)返回一個(gè)基于源DStream的窗口批次計(jì)算后得到的新DStreamcountByWindow(windowLength,slideInterval)返回基于滑動(dòng)窗口的DStream中的元素的數(shù)量reduceByWindow(func,windowLength,slideInterval)基于滑動(dòng)窗口對(duì)源DStream中的元素進(jìn)行聚合操作，得到一個(gè)新的DStreamreduceByKeyAndWindow(func,windowLength,slideInterval,[numTasks])基于滑動(dòng)窗口對(duì)元素為(K,V)鍵值對(duì)的DStream中的值，按K使用func函數(shù)進(jìn)行聚合操作，得到一個(gè)新的DStream（續(xù)表）使用DStream窗口操作方法描述reduceByKeyAndWindow(func,invFunc,windowLength,slideInterval,[numTasks])一個(gè)更高效的reduceByKeyAndWindow()的實(shí)現(xiàn)版本，其中每個(gè)窗口的統(tǒng)計(jì)量是使用前一個(gè)窗口的新數(shù)據(jù)和“反向減少”離開(kāi)窗口的舊數(shù)據(jù)來(lái)實(shí)現(xiàn)的。例如，計(jì)算t+4秒這個(gè)時(shí)刻過(guò)去5秒窗口的WordCount，可以將t+3秒時(shí)刻過(guò)去5秒的統(tǒng)計(jì)量加上[t+3秒,t+4秒]的統(tǒng)計(jì)量，再減去[t?2秒,t?1秒]的統(tǒng)計(jì)量，這種方法可以復(fù)用中間3秒的統(tǒng)計(jì)量，提高統(tǒng)計(jì)的效率countByValueAndWindow(windowLength,slideInterval,[numTasks])基于滑動(dòng)窗口計(jì)算源DStream中每個(gè)RDD內(nèi)每個(gè)元素出現(xiàn)的頻次并返回DStream[(K,Long)]，其中K是RDD中元素的類型，Long是元素頻次。與countByValue一樣，reduce任務(wù)的數(shù)量可以通過(guò)一個(gè)可選參數(shù)進(jìn)行配置以window()為例設(shè)置窗口長(zhǎng)度為3s，滑動(dòng)時(shí)間間隔為1s，截取源DStream中的元素形成新的DStream。使用DStream窗口操作在slave1上啟動(dòng)監(jiān)聽(tīng)，基本上每秒輸入一個(gè)字母，然后取出當(dāng)前時(shí)刻3秒這個(gè)長(zhǎng)度中的所有元素，打印出來(lái)如右圖，從圖中可以看到，到第4秒時(shí)已經(jīng)不到a了，再下一秒就看不到b了，這說(shuō)明此時(shí)a和b已經(jīng)不在當(dāng)前窗口中。使用DStream窗口操作reduceByKeyAndWindow(func,windowLength,slideInterval,[numTasks])操作類似reduceByKey操作，只不過(guò)兩者操作的數(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窗口操作統(tǒng)計(jì)當(dāng)前3秒內(nèi)不同單詞出現(xiàn)的次數(shù)使用DStream窗口操作統(tǒng)計(jì)當(dāng)前3秒內(nèi)不同單詞出現(xiàn)的次數(shù)，運(yùn)行結(jié)果如下圖計(jì)數(shù)。使用DStream窗口操作使用DStream輸出操作DStream輸出操作常用的方法及說(shuō)明方法描述print()在Driver中輸出DStream中數(shù)據(jù)的前10個(gè)元素saveAsTextFiles(prefix,[suffix])將DStream中的內(nèi)容以文本的形式保存為文本文件，其中每次批處理間隔內(nèi)產(chǎn)生的文件再單獨(dú)保存為文件夾，文件夾以prefix_TIME_IN_MS[.suffix]的方式命名saveAsObjectFiles(prefix,[suffix])將DStream中的內(nèi)容按對(duì)象序列化，并且以SequenceFile的格式保存。其中每次批處理間隔內(nèi)產(chǎn)生的文件以prefix_TIME_IN_MS[.suffix]的方式命名saveAsHadoopFiles(prefix,[suffix])將DStream中的內(nèi)容以文本的形式保存為Hadoop文件，其中每次批處理間隔內(nèi)產(chǎn)生的文件以prefix_TIME_IN_MS[.suffix]的方式命名foreachRDD(func)基本的輸出操作，將func函數(shù)應(yīng)用于DStream中的RDD上，輸出數(shù)據(jù)至外部系統(tǒng)，如保存RDD到文件或網(wǎng)絡(luò)數(shù)據(jù)庫(kù)等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輸出操作將nc窗口中輸出的內(nèi)容保存在HDFS的/user/root/saveAsTextFiles文件夾下，設(shè)置每秒生成一個(gè)文件夾。使用DStream輸出操作保存結(jié)果如下圖使用DStream輸出操作foreachRDD是DStream提供的一個(gè)功能強(qiáng)大的方法，它可以將數(shù)據(jù)發(fā)送到外部系統(tǒng)，在使用foreachRDD的過(guò)程中需避免以下錯(cuò)誤。通常將數(shù)據(jù)寫(xiě)入到外部系統(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輸出操作foreachRDD是DStream提供的一個(gè)功能強(qiáng)大的方法，它可以將數(shù)據(jù)發(fā)送到外部系統(tǒng)，在使用foreachRDD的過(guò)程中需避免在SparkDriver端創(chuàng)建連接對(duì)象。針對(duì)以上所說(shuō)的錯(cuò)誤，正確的解決方法是在Worker上創(chuàng)建連接對(duì)象，如代碼所示，但是這種做法又會(huì)引發(fā)另外一種錯(cuò)誤，即為每一個(gè)記錄創(chuàng)建一個(gè)連接對(duì)象。通常，創(chuàng)建一個(gè)連接對(duì)象會(huì)有時(shí)間和資源的開(kāi)銷，因此，為每個(gè)記錄創(chuàng)建和銷毀連接對(duì)象會(huì)導(dǎo)致非常高的開(kāi)銷，減少系統(tǒng)的整體吞吐量。使用DStream輸出操作foreachRDD的正確用法：使用rdd.foreachPartition方法創(chuàng)建一個(gè)單獨(dú)的連接對(duì)象，然后使用該連接對(duì)象輸出所有RDD分區(qū)中的數(shù)據(jù)到外部系統(tǒng)。這不僅可以緩解創(chuàng)建多條記錄連接的開(kāi)銷，還可以通過(guò)在多個(gè)RDDs/batches上重用連接對(duì)象進(jìn)行優(yōu)化。使用DStream輸出操作以網(wǎng)站熱詞排名為例，介紹如何正確使用foreachPartition將處理結(jié)果寫(xiě)到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輸出操作網(wǎng)站熱詞排名在IntelliJIDEA中編寫(xiě)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沒(méi)有提供sort的方法，但是可以使用transform函數(shù)，用RDD的sortByKey實(shí)現(xiàn)。接著需要使用foreachPartition創(chuàng)建MySQL數(shù)據(jù)庫(kù)連接對(duì)象；然后使用該連接對(duì)象輸出數(shù)據(jù)到searchKeyWord表中。使用DStream輸出操作使用DStream輸出操作網(wǎng)站熱詞排名--代碼實(shí)現(xiàn)運(yùn)行程序；在slave1啟動(dòng)監(jiān)聽(tīng)8888端口并輸入數(shù)據(jù)；查看searchKeyWord表中的數(shù)據(jù)；結(jié)果如下圖。使用DStream輸出操作1掌握DStream基礎(chǔ)操作目錄初識(shí)SparkStreaming2實(shí)現(xiàn)書(shū)籍熱度實(shí)時(shí)計(jì)算3掌握了SparkStreaming的DStream編程模型的基礎(chǔ)操作后，即可使用SparkStreaming框架解決實(shí)際的實(shí)時(shí)數(shù)據(jù)流處理問(wèn)題。本節(jié)的任務(wù)如下。使用SparkStreaming實(shí)時(shí)計(jì)算書(shū)籍熱度；根據(jù)書(shū)籍熱度進(jìn)行降序排序，獲取熱度最高的10本圖書(shū)；將最后的結(jié)果寫(xiě)入Hive中。任務(wù)描述將用戶對(duì)書(shū)籍的評(píng)分?jǐn)?shù)據(jù)文件BookRating.txt保存至本地目錄下，設(shè)置每隔60秒隨機(jī)從BookRating.txt文件中挑選100條記錄并添加至新日志文件中，新生成的日志文件存放在“F:\\StreamingData”路徑下。執(zhí)行代碼，產(chǎn)生文件，每個(gè)文件都有100條記錄，如右圖。獲取輸入數(shù)據(jù)源創(chuàng)建rating.scala，實(shí)例化StreamingContext對(duì)象并監(jiān)控“F:\\StreamingData”路徑，實(shí)時(shí)抽取產(chǎn)生的新文件的數(shù)據(jù)并轉(zhuǎn)化為數(shù)據(jù)流，設(shè)置批處理時(shí)間間隔為60秒。在獲取到數(shù)據(jù)流后，通過(guò)split()方法按制表符進(jìn)行切分，并輸出數(shù)據(jù)流進(jìn)行測(cè)試。運(yùn)行結(jié)果如右圖。獲取輸入數(shù)據(jù)源根據(jù)書(shū)籍熱度的計(jì)算公式，需要計(jì)算出用戶的評(píng)分次數(shù)及用戶的平均評(píng)分。首先需要將獲取輸入源的代碼中切分后的數(shù)據(jù)流由DStream形式轉(zhuǎn)換為DataFrame形式，再使用SparkSQLAPI進(jìn)行后續(xù)的數(shù)據(jù)處理。通過(guò)for