SparkStreaming之window滑動窗口應用，Spark Streaming提供了滑動窗口操作的支持，從而讓我們可以對一個滑動窗口內的數據執行計算操作。每次掉落在窗口內的RDD的數據，會被聚合起來執行計算操作，然后生成的RDD，會作為window DStream的一個RDD。
網官圖中所示，就是對每三秒鍾的數據執行一次滑動窗口計算，這3秒內的3個RDD會被聚合起來進行處理，然后過了兩秒鍾，又會對最近三秒內的數據執行滑動窗口計算。所以每個滑動窗口操作，都必須指定兩個參數，窗口長度以及滑動間隔，而且這兩個參數值都必須是batch間隔的整數倍。
Spark Streaming對滑動窗口的支持，是比Storm更加完善和強大的。
之前有些朋友問：
spark官網圖片中： 滑動窗口寬度是3個時間單位，滑動時間是2兩個單位，這樣的話中間time3的Dstream不是重復計算了嗎？
Answer：比如下面這個例子是針對熱搜的應用場景，官方的例子也可能是是針對不同的場景給出了的。如果你不想出現重疊的部分，把滑動間隔由2改成3即可
SparkStreaming對滑動窗口支持的轉換操作：
示例講解：
1、window(windowLength, slideInterval)
該操作由一個DStream對象調用，傳入一個窗口長度參數，一個窗口移動速率參數，然后將當前時刻當前長度窗口中的元素取出形成一個新的DStream。
下面的代碼以長度為3，移動速率為1截取源DStream中的元素形成新的DStream。
val windowWords = words.window(Seconds( 3 ), Seconds( 1))
基本上每秒輸入一個字母，然后取出當前時刻3秒這個長度中的所有元素，打印出來。從上面的截圖中可以看到，下一秒時已經看不到a了，再下一秒，已經看不到b和c了。表示a, b, c已經不在當前的窗口中。
2、 countByWindow(windowLength,slideInterval)
返回指定長度窗口中的元素個數。
代碼如下，統計當前3秒長度的時間窗口的DStream中元素的個數：
val windowWords = words.countByWindow(Seconds( 3 ), Seconds( 1))
3、 reduceByWindow(func, windowLength,slideInterval)
類似於上面的reduce操作，只不過這里不再是對整個調用DStream進行reduce操作，而是在調用DStream上首先取窗口函數的元素形成新的DStream，然后在窗口元素形成的DStream上進行reduce。
val windowWords = words.reduceByWindow(_ + "-" + _, Seconds( 3) , Seconds( 1 ))
4、 reduceByKeyAndWindow(func,windowLength, slideInterval, [numTasks])
調用該操作的DStream中的元素格式為(k, v)，整個操作類似於前面的reduceByKey，只不過對應的數據源不同，reduceByKeyAndWindow的數據源是基於該DStream的窗口長度中的所有數據。該操作也有一個可選的並發數參數。
下面代碼中，將當前長度為3的時間窗口中的所有數據元素根據key進行合並，統計當前3秒中內不同單詞出現的次數。
val windowWords = pairs.reduceByKeyAndWindow((a:Int , b:Int) => (a + b) , Seconds(3 ) , Seconds( 1 ))
5、 reduceByKeyAndWindow(func, invFunc,windowLength, slideInterval, [numTasks])
這個窗口操作和上一個的區別是多傳入一個函數invFunc。前面的func作用和上一個reduceByKeyAndWindow相同，后面的invFunc是用於處理流出rdd的。
在下面這個例子中，如果把3秒的時間窗口當成一個池塘，池塘每一秒都會有魚游進或者游出，那么第一個函數表示每由進來一條魚，就在該類魚的數量上累加。而第二個函數是，每由出去一條魚，就將該魚的總數減去一。
val windowWords = pairs.reduceByKeyAndWindow((a: Int, b:Int ) => (a + b) , (a:Int, b: Int) => (a - b) , Seconds( 3 ), Seconds( 1 ))
下面是演示結果，最終的結果是該3秒長度的窗口中歷史上出現過的所有不同單詞個數都為0。
6、 countByValueAndWindow(windowLength,slideInterval, [numTasks])
類似於前面的countByValue操作，調用該操作的DStream數據格式為(K, v)，返回的DStream格式為(K, Long)。統計當前時間窗口中元素值相同的元素的個數。
val windowWords = words.countByValueAndWindow(Seconds( 3 ), Seconds( 1))
示例二：熱點搜索詞滑動統計，每隔10秒鍾，統計最近60秒鍾的搜索詞的搜索頻次，並打印出排名最靠前的3個搜索詞以及出現次數
Scala版本：
package com.spark.streaming import org.apache.spark.streaming.Seconds import org.apache.spark.streaming.StreamingContext import org.apache.spark.SparkConf /** * @author Ganymede */ object WindowHotWordS { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("WindowHotWordS").setMaster("local[2]") //Scala中，創建的是StreamingContext val ssc = new StreamingContext(conf, Seconds(5)) val searchLogsDStream = ssc.socketTextStream("spark1", 9999) val searchWordsDStream = searchLogsDStream.map { searchLog => searchLog.split(" ")(1) } val searchWordPairDStream = searchWordsDStream.map { searchWord => (searchWord, 1) } // reduceByKeyAndWindow // 第二個參數，是窗口長度，這是是60秒 // 第三個參數，是滑動間隔，這里是10秒 // 也就是說，每隔10秒鍾，將最近60秒的數據，作為一個窗口，進行內部的RDD的聚合，然后統一對一個RDD進行后續計算 // 而是只是放在那里 // 然后，等待我們的滑動間隔到了以后，10秒到了，會將之前60秒的RDD，因為一個batch間隔是5秒，所以之前60秒，就有12個RDD，給聚合起來，然后統一執行reduceByKey操作 // 所以這里的reduceByKeyAndWindow，是針對每個窗口執行計算的，而不是針對 某個DStream中的RDD // 每隔10秒鍾，出來 之前60秒的收集到的單詞的統計次數 val searchWordCountsDStream = searchWordPairDStream.reduceByKeyAndWindow((v1: Int, v2: Int) => v1 + v2, Seconds(60), Seconds(10)) val finalDStream = searchWordCountsDStream.transform(searchWordCountsRDD => { val countSearchWordsRDD = searchWordCountsRDD.map(tuple => (tuple._2, tuple._1)) val sortedCountSearchWordsRDD = countSearchWordsRDD.sortByKey(false) val sortedSearchWordCountsRDD = sortedCountSearchWordsRDD.map(tuple => (tuple._1, tuple._2)) val top3SearchWordCounts = sortedSearchWordCountsRDD.take(3) for (tuple <- top3SearchWordCounts) { println("result : " + tuple) } searchWordCountsRDD }) finalDStream.print() ssc.start() ssc.awaitTermination() } }