大數(shù)據(jù)分析技術(shù)與實(shí)戰(zhàn)之 Spark Streaming

Spark是基于內(nèi)存的大數(shù)據(jù)綜合處理引擎，具有優(yōu)秀的作業(yè)調(diào)度機(jī)制和快速的分布式計(jì)算能力，使其能夠更加高效地進(jìn)行迭代計(jì)算，因此Spark能夠在一定程度上實(shí)現(xiàn)大數(shù)據(jù)的流式處理。
隨著信息技術(shù)的迅猛發(fā)展，數(shù)據(jù)量呈現(xiàn)出爆炸式增長(zhǎng)趨勢(shì)，數(shù)據(jù)的種類與變化速度也遠(yuǎn)遠(yuǎn)超出人們的想象，因此人們對(duì)大數(shù)據(jù)處理提出了更高的要求，越來(lái)越多的領(lǐng)域迫切需要大數(shù)據(jù)技術(shù)來(lái)解決領(lǐng)域內(nèi)的關(guān)鍵問(wèn)題。在一些特定的領(lǐng)域中(例如金融、災(zāi)害預(yù)警等)，時(shí)間就是金錢、時(shí)間可能就是生命!然而傳統(tǒng)的批處理框架卻一直難以滿足這些領(lǐng)域中的實(shí)時(shí)性需求。為此，涌現(xiàn)出了一批如S4、Storm的流式計(jì)算框架。Spark是基于內(nèi)存的大數(shù)據(jù)綜合處理引擎，具有優(yōu)秀的作業(yè)調(diào)度機(jī)制和快速的分布式計(jì)算能力，使其能夠更加高效地進(jìn)行迭代計(jì)算，因此Spark能夠在一定程度上實(shí)現(xiàn)大數(shù)據(jù)的流式處理。
Spark Streaming是Spark上的一個(gè)流式處理框架，可以面向海量數(shù)據(jù)實(shí)現(xiàn)高吞吐量、高容錯(cuò)的實(shí)時(shí)計(jì)算。Spark Streaming支持多種類型數(shù)據(jù)源，包括Kafka、Flume、trwitter、zeroMQ、Kinesis以及TCP sockets等，如圖1所示。Spark Streaming實(shí)時(shí)接收數(shù)據(jù)流，并按照一定的時(shí)間間隔將連續(xù)的數(shù)據(jù)流拆分成一批批離散的數(shù)據(jù)集;然后應(yīng)用諸如map、reducluce、join和window等豐富的API進(jìn)行復(fù)雜的數(shù)據(jù)處理;最后提交給Spark引擎進(jìn)行運(yùn)算，得到批量結(jié)果數(shù)據(jù)，因此其也被稱為準(zhǔn)實(shí)時(shí)處理系統(tǒng)。

圖1 Spark Streaming支持多種類型數(shù)據(jù)源
目前應(yīng)用最廣泛的×××式處理框架是Storm。Spark Streaming 最低0.5~2s做一次處理(而Storm最快可達(dá)0.1s)，在實(shí)時(shí)性和容錯(cuò)方面不如Storm。然而Spark Streaming的集成性非常好，通過(guò)RDD不僅能夠與Spark上的所有組件無(wú)縫銜接共享數(shù)據(jù)，還能非常容易地與Kafka、Flume等分布式日志收集框架進(jìn)行集成;同時(shí)Spark Streaming的吞吐量非常高，遠(yuǎn)遠(yuǎn)優(yōu)于Storm的吞吐量，如圖2所示。所以雖然Spark Streaming的處理延遲高于Storm，但是在集成性與吞吐量方面的優(yōu)勢(shì)使其更適用于大數(shù)據(jù)背景。

Spark Streaming基礎(chǔ)概念
批處理時(shí)間間隔
在Spark Streaming中，對(duì)數(shù)據(jù)的采集是實(shí)時(shí)、逐條進(jìn)行的，但是對(duì)數(shù)據(jù)的處理卻是分批進(jìn)行的。因此，Spark Streaming需要設(shè)定一個(gè)時(shí)間間隔，將該時(shí)間間隔內(nèi)采集到的數(shù)據(jù)統(tǒng)一進(jìn)行處理，這個(gè)間隔稱為批處理時(shí)間間隔。
也就是說(shuō)對(duì)于源源不斷的數(shù)據(jù)，Spark Streaming是通過(guò)切分的方式，先將連續(xù)的數(shù)據(jù)流進(jìn)行離散化處理。數(shù)據(jù)流每被切分一次，對(duì)應(yīng)生成一個(gè)RDD，每個(gè)RDD都包含了一個(gè)時(shí)間間隔內(nèi)所獲取到的所有數(shù)據(jù)，因此數(shù)據(jù)流被轉(zhuǎn)換為由若干個(gè)RDD構(gòu)成的有序集合，而批處理時(shí)間間隔決定了Spark Streaming需要多久對(duì)數(shù)據(jù)流切分一次。Spark Streaming是Spark上的組件，其獲取的數(shù)據(jù)和數(shù)據(jù)上的操作最終仍以Spark作業(yè)的形式在底層的Spark內(nèi)核中進(jìn)行計(jì)算，因此批處理時(shí)間間隔不僅影響數(shù)據(jù)處理的吞吐量，同時(shí)也決定了Spark Streaming向Spark提交作業(yè)的頻率和數(shù)據(jù)處理的延遲。需要注意的是，批處理時(shí)間間隔的設(shè)置會(huì)伴隨Spark Streaming應(yīng)用程序的整個(gè)生命周期，無(wú)法在程序運(yùn)行期間動(dòng)態(tài)修改，所以需要綜合考慮實(shí)際應(yīng)用場(chǎng)景中的數(shù)據(jù)流特點(diǎn)和集群的處理性能等多種因素進(jìn)行設(shè)定。
窗口時(shí)間間隔
窗口時(shí)間間隔又稱為窗口長(zhǎng)度，它是一個(gè)抽象的時(shí)間概念，決定了Spark Streaming對(duì)RDD序列進(jìn)行處理的范圍與粒度，即用戶可以通過(guò)設(shè)置窗口長(zhǎng)度來(lái)對(duì)一定時(shí)間范圍內(nèi)的數(shù)據(jù)進(jìn)行統(tǒng)計(jì)和分析。如果設(shè)批處理時(shí)間設(shè)為1s，窗口時(shí)間間隔為3s，如3圖所示，其中每個(gè)實(shí)心矩形表示Spark Streaming每1秒鐘切分出的一個(gè)RDD，若干個(gè)實(shí)心矩形塊表示一個(gè)以時(shí)間為序的RDD序列，而透明矩形框表示窗口時(shí)間間隔。易知窗口內(nèi)RDD的數(shù)量最多為3個(gè)，即Spark Streming 每次最多對(duì)3個(gè)RDD中的數(shù)據(jù)進(jìn)行統(tǒng)計(jì)和分析。對(duì)于窗口時(shí)間間隔還需要注意以下幾點(diǎn)：
以圖3為例，在系統(tǒng)啟動(dòng)后的前3s內(nèi)，因進(jìn)入窗口的RDD不足3個(gè)，但是隨著時(shí)間的推移，最終窗口將被填滿。
不同窗口內(nèi)所包含的RDD可能會(huì)有重疊，即當(dāng)前窗口內(nèi)的數(shù)據(jù)可能被其后續(xù)若干個(gè)窗口所包含，因此在一些應(yīng)用場(chǎng)景中，對(duì)于已經(jīng)處理過(guò)的數(shù)據(jù)不能立即刪除，以備后續(xù)計(jì)算使用。
窗口時(shí)間間隔必須是批處理時(shí)間間隔的整數(shù)倍。

圖3 窗口時(shí)間間隔示意圖
滑動(dòng)時(shí)間間隔
滑動(dòng)時(shí)間間隔決定了Spark Streaming對(duì)數(shù)據(jù)進(jìn)行統(tǒng)計(jì)與分析的頻率，多出現(xiàn)在與窗口相關(guān)的操作中?；瑒?dòng)時(shí)間間隔是基于批處理時(shí)間間隔提出的，其必須是批處理時(shí)間間隔的整數(shù)倍。在默認(rèn)的情況下滑動(dòng)時(shí)間間隔設(shè)置為與批處理時(shí)間間隔相同的值。如果批處理時(shí)間間隔為1s，窗口間隔為3s，滑動(dòng)時(shí)間間隔為2s，如圖4所示，其含義是每隔2s對(duì)過(guò)去3s內(nèi)產(chǎn)生的3個(gè)RDD進(jìn)行統(tǒng)計(jì)分析。

圖4 滑動(dòng)時(shí)間間隔、窗口時(shí)間間隔、批處理時(shí)間間隔綜合示意圖
DStream基本概念
DStream是Spark Streaming的一個(gè)基本抽象，它以離散化的RDD序列的形式近似描述了連續(xù)的數(shù)據(jù)流。DStream本質(zhì)上是一個(gè)以時(shí)間為鍵，RDD為值的哈希表，保存了按時(shí)間順序產(chǎn)生的RDD，而每個(gè)RDD封裝了批處理時(shí)間間隔內(nèi)獲取到的數(shù)據(jù)。Spark Streaming每次將新產(chǎn)生的RDD添加到哈希表中，而對(duì)于已經(jīng)不再需要的RDD則會(huì)從這個(gè)哈希表中刪除，所以DStream也可以簡(jiǎn)單地理解為以時(shí)間為鍵的RDD的動(dòng)態(tài)序列。設(shè)批處理時(shí)間間隔為1s，圖5為4s內(nèi)產(chǎn)生的DStream示意圖。

Spark Streaming編程模式與案例分析
Spark Streaming編程模式
下面以Spark Streaming官方提供的WordCount代碼為例來(lái)介紹Spark Streaming的使用方式。
示例1：

Spark Streaming應(yīng)用程序在功能結(jié)構(gòu)上通常包含以下五部分，如上述示例1所示。
導(dǎo)入Spark Streaming相關(guān)包：Spark Streaming作為Spark框架上的一個(gè)組件，具有很好的集成性。在開發(fā)Spark Streaming應(yīng)用程序時(shí)，只需導(dǎo)入Spark Streaming相關(guān)包，無(wú)需額外的參數(shù)配置。
創(chuàng)建StreamingContext對(duì)象：同Spark應(yīng)用程序中的SparkContext對(duì)象一樣， StreamingContext對(duì)象是Spark Streaming應(yīng)用程序與集群進(jìn)行交互的唯一通道，其中封裝了Spark集群的環(huán)境信息和應(yīng)用程序的一些屬性信息。在該對(duì)象中通常需要指明應(yīng)用程序的運(yùn)行模式(示例1中設(shè)為local[2])、設(shè)定應(yīng)用程序名稱(示例1中設(shè)為NetworkWordCount)、設(shè)定批處理時(shí)間間隔(示例1中設(shè)為Seconds(1)即1秒鐘)，其中批處理時(shí)間間隔需要根據(jù)用戶的需求和集群的處理能力進(jìn)行適當(dāng)?shù)卦O(shè)置。
創(chuàng)建InputDStream：Spark Streaming需要根據(jù)數(shù)據(jù)源類型選擇相應(yīng)的創(chuàng)建DStream的方法。示例1中Spark Streaming通過(guò)StreamingContext對(duì)象調(diào)用socketTextStream方法處理以socket連接類型數(shù)據(jù)源，創(chuàng)建出DStream即lines。Spark Streaming同時(shí)支持多種不同的數(shù)據(jù)源類型，其中包括Kafka、Flume、HDFS/S3、Kinesis和Twitter等數(shù)據(jù)源。
操作DStream：對(duì)于從數(shù)據(jù)源得到的DStream，用戶可以調(diào)用豐富的操作對(duì)其進(jìn)行處理。示例1中針對(duì)lines的一系列操作就是一個(gè)典型的WordCount執(zhí)行流程：對(duì)于當(dāng)前批處理時(shí)間間隔內(nèi)的文本數(shù)據(jù)以空格進(jìn)行切分，進(jìn)而得到words;再將words中每個(gè)單詞轉(zhuǎn)換為二元組，進(jìn)而得到pairs;最后利用reduceByKey方法進(jìn)行統(tǒng)計(jì)。
啟動(dòng)與停止Spark Streaming應(yīng)用程序：在啟動(dòng)Spark Streaming應(yīng)用程序之前，DStream上所有的操作僅僅是定義了數(shù)據(jù)的處理流程，程序并沒(méi)有真正連接上數(shù)據(jù)源，也沒(méi)有對(duì)數(shù)據(jù)進(jìn)行任何操作，當(dāng)ssc.start()啟動(dòng)后程序中定義的操作才會(huì)真正開始執(zhí)行。
文本文件數(shù)據(jù)處理案例
功能需求
實(shí)時(shí)監(jiān)聽并獲取本地home/dong/Streamingtext目錄中新生成的文件(文件均為英文文本文件，單詞之間使用空格進(jìn)行間隔)，并對(duì)文件中各單詞出現(xiàn)的次數(shù)進(jìn)行統(tǒng)計(jì)。
代碼實(shí)現(xiàn)

運(yùn)行演示
第1步，啟動(dòng)Hadoop與Spark。

第2步，創(chuàng)建Streaming監(jiān)控目錄。

在dong用戶主目錄下創(chuàng)建Streamingtext為Spark Streaming監(jiān)控的目錄，創(chuàng)建后如圖6所示。

圖6 dong用戶主目錄下創(chuàng)建Streamingtext文件夾
第3步，在IntelliJ IDEA中編輯運(yùn)行Streaming程序。在IntelliJ IDEA中創(chuàng)建工程StreamingFileWordCount，編輯對(duì)象StreamingFileWordCount，如圖7所示。

圖7 IntelliJ IDEA中StreamingFileWordCount示意圖
由于該示例沒(méi)有輸入?yún)?shù)，因此不需要配置參數(shù)，可直接單擊右鍵->單擊"Run‘StreamingFileWordCount’ "。
第4步，在監(jiān)聽目錄下創(chuàng)建文本文件。在master節(jié)點(diǎn)上的/home/dong/Streamingtext中分別創(chuàng)建file1.txt與file2.txt。
file1.txt內(nèi)容如下：

file2.txt內(nèi)容如下：

創(chuàng)建后，/home/dong/Streamingtext中內(nèi)容如圖8所示。

圖8 Streamingtext文件夾內(nèi)容示意圖
查看結(jié)果
終端窗口輸出了每個(gè)批處理時(shí)間間隔(20秒)內(nèi)，/home/dong/Streamingtext中新生成文件所包含的各單詞個(gè)數(shù)，如圖9所示。

圖9 StreamingFileWordCount運(yùn)行結(jié)果示意圖
網(wǎng)絡(luò)數(shù)據(jù)處理案例
功能需求
監(jiān)聽本地節(jié)點(diǎn)指定端口傳輸?shù)臄?shù)據(jù)流(本案例為master節(jié)點(diǎn)9999端口的英文文本數(shù)據(jù)，以逗號(hào)間隔單詞)，每5秒統(tǒng)計(jì)一次該時(shí)間間隔內(nèi)收集到的各單詞的個(gè)數(shù)。
代碼實(shí)現(xiàn)
本案例涉及數(shù)據(jù)流模擬器和分析器兩部分。為了更接近真實(shí)的網(wǎng)絡(luò)環(huán)境，首先定義數(shù)據(jù)流模擬器，該模擬器以Socket方式監(jiān)聽網(wǎng)絡(luò)中指定節(jié)點(diǎn)上的指定端口號(hào)(master節(jié)點(diǎn)9999端口)，當(dāng)外部程序通過(guò)該端口連接并請(qǐng)求數(shù)據(jù)時(shí)，數(shù)據(jù)流模擬器將定時(shí)地從指定文本文件中隨機(jī)選取數(shù)據(jù)發(fā)送至指定端口(每間隔1秒鐘數(shù)據(jù)流模擬器從master節(jié)點(diǎn)上的/home/dong/Streamingtext/file1.txt中隨機(jī)截取一行文本發(fā)送給master節(jié)點(diǎn)的9999端口)，通過(guò)這種方式模擬網(wǎng)絡(luò)環(huán)境下源源不斷的數(shù)據(jù)流。針對(duì)獲取到的實(shí)時(shí)數(shù)據(jù)，再定義分析器(Spark Streaming應(yīng)用程序)，用以統(tǒng)計(jì)時(shí)間間隔(5秒)內(nèi)收集到的單詞個(gè)數(shù)。
數(shù)據(jù)流模擬器代碼實(shí)現(xiàn)如下：

分析器代碼如下：

運(yùn)行演示
第1步，在IntelliJ IDEA中編輯運(yùn)行Streaming程序。master節(jié)點(diǎn)啟動(dòng)IntelliJ IDEA，創(chuàng)建工程N(yùn)etworkWordCount，編輯模擬器與分析器。模擬器如圖10所示，分析器如圖11所示。

圖10 IntelliJ IDEA中數(shù)據(jù)流模擬器示意圖

圖11 IntelliJ IDEA中分析器示意圖
第2步，創(chuàng)建模擬器數(shù)據(jù)源文件。在master節(jié)點(diǎn)創(chuàng)建/home/dong/Streamingtext目錄，在其中創(chuàng)建文本文件file1.txt。
file1.txt內(nèi)容如下：

第3步，打包數(shù)據(jù)流模擬器。打包過(guò)程詳見(jiàn)本書4.3.3節(jié)。在Artifacts打包配置界面中，根據(jù)用戶實(shí)際scala安裝目錄，在Class Path中添加下述scala依賴包，如圖12所示。

圖12 在Class Path中添加scala依賴包
打包后在主目錄下生成NetworkWordCount.jar，如圖13所示。
圖13 在dong用戶主目錄下生成NetworkWordCount.jar示意圖
第4步，啟動(dòng)數(shù)據(jù)流模擬器。在master節(jié)點(diǎn)開啟控制終端，通過(guò)下面代碼啟動(dòng)數(shù)據(jù)流模擬器。

數(shù)據(jù)流模擬器每間隔1000毫秒從/home/dong/Streamingtext/file1.txt中隨機(jī)截取一行文本發(fā)送給master節(jié)點(diǎn)的9999端口。在分析器未連接時(shí)，數(shù)據(jù)流模擬器處于阻塞狀態(tài)，終端不會(huì)顯示輸出的文本。
第5步，運(yùn)行分析器。在master上啟動(dòng)IntelliJ IDEA編寫分析器代碼，然后單擊菜單"Build->"Build Artifacts"，通過(guò)Application選項(xiàng)配置分析器運(yùn)行所需的參數(shù)，其中Socket主機(jī)名為master、端口號(hào)為9999，參數(shù)之間用空格間隔，如圖13所示。

圖13 分析器參數(shù)配置示意圖
配置好參數(shù)后返回IntelliJ IDEA菜單欄，單擊"Run"->"Build Artifacts"運(yùn)行分析器。
查看結(jié)果
第1步，在master上查看數(shù)據(jù)流模擬器運(yùn)行情況。IntelliJ IDEA運(yùn)行分析器從而與數(shù)據(jù)流模擬器建立連接。當(dāng)檢測(cè)到外部連接時(shí)，數(shù)據(jù)流模擬器將每隔1000毫秒從/home/dong/Streamingtext/file1.txt中隨機(jī)截取一行文本發(fā)送給master節(jié)點(diǎn)上的9999端口。為方便講解和說(shuō)明，file1.txt中每一行只包含一個(gè)單詞，因此數(shù)據(jù)流模擬器每次僅發(fā)送一個(gè)單詞給端口，如圖14所示。

圖14 在master上模擬器運(yùn)行結(jié)果
第2步，在master的IntelliJ IDEA中查看分析器運(yùn)行情況。在IntelliJ IDEA的運(yùn)行日志窗口中，可以觀察到統(tǒng)計(jì)結(jié)果。通過(guò)分析可知Spark Streaming每個(gè)批處理時(shí)間間隔內(nèi)獲取的單詞數(shù)為5，剛好是5秒內(nèi)發(fā)送單詞的總數(shù)，并對(duì)各單詞進(jìn)行了統(tǒng)計(jì)，如圖15所示。

stateful應(yīng)用案例
在很多數(shù)據(jù)流相關(guān)的實(shí)際應(yīng)用場(chǎng)景中，對(duì)當(dāng)前數(shù)據(jù)的統(tǒng)計(jì)分析需要借助于先前的數(shù)據(jù)處理結(jié)果完成。例如電商每間隔10分鐘統(tǒng)計(jì)某一商品當(dāng)前累計(jì)銷售總額、車站每隔3小時(shí)統(tǒng)計(jì)當(dāng)前客流總量，等等。此類應(yīng)用需求可借助于Spark Streaming的有狀態(tài)轉(zhuǎn)換操作實(shí)現(xiàn)。
功能需求
監(jiān)聽網(wǎng)絡(luò)中某節(jié)點(diǎn)上指定端口傳輸?shù)臄?shù)據(jù)流(slave1節(jié)點(diǎn)9999端口的英文文本數(shù)據(jù)，以逗號(hào)間隔單詞)，每5秒分別統(tǒng)計(jì)各單詞的累計(jì)出現(xiàn)次數(shù)。
代碼實(shí)現(xiàn)
本案例功能的實(shí)現(xiàn)涉及數(shù)據(jù)流模擬器和分析器兩部分。
分析器代碼：

運(yùn)行演示
第1步，slave1節(jié)點(diǎn)啟動(dòng)數(shù)據(jù)流模擬器。
第2步，打包分析器。master節(jié)點(diǎn)啟動(dòng)IntelliJ IDEA創(chuàng)建工程StatefulWordCount編輯分析器，如圖16所示，并將分析器直接打包至master節(jié)點(diǎn)dong用戶的主目錄下，如圖17所示。

圖16 IntelliJ IDEA中StatefulWordCount示意圖

圖17 master上的StatefulWordCount.jar示意圖
第3步，運(yùn)行分析器。在master節(jié)點(diǎn)開啟終端，通過(guò)下面代碼向Spark集群提交應(yīng)用程序。

查看結(jié)果
第1步，查看slave1上數(shù)據(jù)流模擬器運(yùn)行情況。分析器在集群上提交運(yùn)行后與slave1上運(yùn)行的數(shù)據(jù)流模擬器建立連接。當(dāng)檢測(cè)到外部連接時(shí)，數(shù)據(jù)流模擬器將每隔1000毫秒從/home/dong/Streamingtext/file1.txt中隨機(jī)截取一行文本發(fā)送給slave1節(jié)點(diǎn)上的9999端口。由于該文本文件中每一行只包含一個(gè)單詞，因此每秒僅發(fā)送一個(gè)單詞給端口。如圖18所示。

圖18 slave1上數(shù)據(jù)流模擬器運(yùn)行示意圖
第2步，查看master上分析器運(yùn)行情況。在master節(jié)點(diǎn)的提交窗口中可以查看到統(tǒng)計(jì)結(jié)果，如圖19所示。

圖19 master上分析器運(yùn)行示意圖
圖中表明截至147920770500ms分析器共接收到14個(gè)單詞，其中"spark"累計(jì)出現(xiàn)3次，"hbase"累計(jì)出現(xiàn)5次，"hello"累計(jì)出現(xiàn)3次，"world"累計(jì)出現(xiàn)3次。由于批處理時(shí)間間隔是5s，模擬器每1秒發(fā)送1個(gè)單詞，使得分析器在5s內(nèi)共接收到5個(gè)單詞，因此截止至147920771000ms，分析器共收到19個(gè)單詞，其中"spark"累計(jì)出現(xiàn)5次，"hbase"累計(jì)出現(xiàn)7次，"hello"累計(jì)出現(xiàn)4次，"world"累計(jì)出現(xiàn)3次。
第3步，查看HDFS中持久化目錄。運(yùn)行后查看HDFS上的持久化目錄/user/dong/input/StatefulWordCountlog，如圖20所示。Streaming應(yīng)用程序?qū)⒔邮盏降木W(wǎng)絡(luò)數(shù)據(jù)持久化至該目錄下，便于容錯(cuò)處理。

圖20 HDFS上持久化目錄示意圖
window應(yīng)用案例
在實(shí)際生產(chǎn)環(huán)境中，與窗口相關(guān)的應(yīng)用場(chǎng)景很常見(jiàn)，例如電商每間隔10分鐘小時(shí)統(tǒng)計(jì)某一商品前30分鐘內(nèi)累計(jì)銷售總額、車站每隔1小時(shí)統(tǒng)計(jì)前3個(gè)小時(shí)內(nèi)的客流量等，此類需求可借助Spark Streaming中的window相關(guān)操作實(shí)現(xiàn)。window應(yīng)用案例同時(shí)涉及批處理時(shí)間間隔、窗口時(shí)間間隔與滑動(dòng)時(shí)間間隔。
功能需求
監(jiān)聽網(wǎng)絡(luò)中某節(jié)點(diǎn)上指定端口傳輸?shù)臄?shù)據(jù)流(slave1節(jié)點(diǎn)上9999端口的英文文本數(shù)據(jù)，以逗號(hào)間隔單詞)，每10秒統(tǒng)計(jì)前30秒各單詞累計(jì)出現(xiàn)的次數(shù)。
代碼實(shí)現(xiàn)
本例功能的實(shí)現(xiàn)涉及數(shù)據(jù)流模擬器和分析器兩部分。
分析器代碼：

運(yùn)行演示 第1步，slave1節(jié)點(diǎn)啟動(dòng)數(shù)據(jù)流模擬器。 第2步，打包分析器。在master節(jié)點(diǎn)啟動(dòng)IntelliJ IDEA創(chuàng)建工程WindowWordCount編輯分析器，如圖21，并將分析器直接打包至master節(jié)點(diǎn)dong用戶的主目錄下，如圖22所示。

圖21 IntelliJ IDEA中WindowWordCount示意圖

圖22 master上WindowWordCount.jar示意圖
第3步，運(yùn)行分析器。在master節(jié)點(diǎn)開啟終端，通過(guò)下面代碼向Spark集群提交應(yīng)用程序。

查看結(jié)果
第1步 在slave1上查看數(shù)據(jù)流模擬器運(yùn)行情況。分析器在集群上提交運(yùn)行后與slave1上運(yùn)行的數(shù)據(jù)流模擬器建立連接。當(dāng)檢測(cè)到外部連接時(shí)，數(shù)據(jù)流模擬器將每隔1000毫秒從/home/dong/Streamingtext/file1.txt中隨機(jī)截取一行文本發(fā)送給slave1節(jié)點(diǎn)的9999端口。由于該文本文件中每一行只包含一個(gè)單詞和一個(gè)逗號(hào)，因此每秒僅發(fā)送一個(gè)單詞和一個(gè)逗號(hào)給端口，如圖23所示。

圖23 在slave1上數(shù)據(jù)流模擬器運(yùn)行示意圖
第2步，在master上查看分析器運(yùn)行情況。在master節(jié)點(diǎn)的提交窗口中可以查看到統(tǒng)計(jì)結(jié)果。在WindowWordCount應(yīng)用程序啟動(dòng)初期，窗口并沒(méi)有被接收到的單詞填滿，但隨著時(shí)間的推進(jìn)，每個(gè)窗口中的單詞數(shù)目最終固定為30個(gè)。圖7.35只是截取了運(yùn)行結(jié)果中的三個(gè)批次。由于設(shè)定了窗口時(shí)間間隔是30s，滑動(dòng)時(shí)間間隔是10s，且數(shù)據(jù)流模擬器每間隔1s發(fā)送一個(gè)單詞，因此WindowWordCount每間隔10s對(duì)過(guò)去30s內(nèi)收到的各單詞個(gè)數(shù)進(jìn)行統(tǒng)計(jì)。圖24中截至1479276925000ms分析器對(duì)過(guò)去30s內(nèi)收到的30個(gè)單詞進(jìn)行統(tǒng)計(jì)，其中"spark"累計(jì)出現(xiàn)5次，"hbase"累計(jì)出現(xiàn)8次，"hello"累計(jì)出現(xiàn)9次，"world"累計(jì)出現(xiàn)8次。再間隔10s，截至1479276935000ms，分析器對(duì)過(guò)去30s內(nèi)收到的30個(gè)單詞進(jìn)行統(tǒng)計(jì)，其中"spark"累計(jì)出現(xiàn)8次，"hbase"累計(jì)出現(xiàn)9次，"hello"累計(jì)出現(xiàn)7次，"world"累計(jì)出現(xiàn)6次。

圖24 在master上分析器運(yùn)行示意圖
第3步，查看持久化數(shù)據(jù)。運(yùn)行后查看HDFS上的持久化目錄/user/dong/input/WindowWordCountlog，如圖25所示。Streaming應(yīng)用程序?qū)⒔邮盏降木W(wǎng)絡(luò)數(shù)據(jù)持久化至該目錄下，便于容錯(cuò)處理。