說到流處理，Spark為我們提供了窗口函數(shù)，允許在滑動數(shù)據窗口上應用轉換，常用場景如每五分鐘商場人流密度、每分鐘流量等等，接下來我們通過畫圖來了解Spark Streaming的窗口函數(shù)如何工作的，處理過程圖如下所示：

上圖中綠色的小框框是一批一批的數(shù)據流，虛線框和實線框分別是前一個窗口和后一個窗口，從圖中可以看出后一個窗口在前一個窗口基礎上移動了兩個批次的數(shù)據流，而我們真正通過算子操作的數(shù)據其實就是窗口內所有的數(shù)據流。

在代碼實現(xiàn)前了解下窗口操作常用的函數(shù)有：

• window
• countByWindow
• reduceByWindow
• reduceByKeyAndWindow
• reduceByKeyAndWindow
• countByValueAndWindow

window最原始的窗口，提供兩個參數(shù)，第一個參數(shù)是窗口長度，第二個參數(shù)是滑動間隔，返回一個新的DStream， 返回的結果可以進行算子操作，代碼實現(xiàn)如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == '__main__':
    sc = SparkContext(appName="windowStream", master="local[*]")
    # 第二個參數(shù)指統(tǒng)計多長時間的數(shù)據
    ssc = StreamingContext(sc, 5)
    lines = ssc.socketTextStream("localhost", 9999)
    # 第一個參數(shù)是窗口長度，這里是60秒， 第二個參數(shù)是滑動間隔，這里是10秒
    dstream = lines.window(60, 10)
    dstream.pprint()
    # -------------------------------------------
    # Time: 2019-08-13 19:46:45
    # -------------------------------------------
    # hello
    # world
    ssc.start()
    ssc.awaitTermination()
            
          

現(xiàn)在終端使用nc發(fā)送數(shù)據

root@root:~$ nc -lk 9999
hello
world

countByWindow統(tǒng)計每個滑動窗口內數(shù)據條數(shù)，要注意的是使用該函數(shù)要加上checkpoint機制，代碼實現(xiàn)如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == '__main__':
    sc = SparkContext(appName="windowStream", master="local[*]")
    # 第二個參數(shù)指統(tǒng)計多長時間的數(shù)據
    ssc = StreamingContext(sc, 5)
    ssc.checkpoint("/tmp/window")
    lines = ssc.socketTextStream("localhost", 9999)
    # 第一個參數(shù)是窗口長度，這里是60秒， 第二個參數(shù)是滑動間隔，這里是10秒
    dstream = lines.countByWindow(60, 10)
    dstream.pprint()
    # -------------------------------------------
    # Time: 2019-08-14 18:56:40
    # -------------------------------------------
    # 2
    ssc.start()
    ssc.awaitTermination()
            
          

reduceByWindow聚合每個鍵的值，底層執(zhí)行的是reduceByKeyAndWindow，實現(xiàn)代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

def fun(x):
    return x

if __name__ == '__main__':
    sc = SparkContext(appName="windowStream", master="local[*]")
    # 第二個參數(shù)指統(tǒng)計多長時間的數(shù)據
    ssc = StreamingContext(sc, 5)
    ssc.checkpoint("/tmp/window")
    lines = ssc.socketTextStream("localhost", 9999)
    # 第一個參數(shù)執(zhí)行指定函數(shù)， 第二個參數(shù)是窗口長度，這里是60秒， 第三個參數(shù)是滑動間隔，這里是10秒
    dstream = lines.reduceByWindow(fun, 60, 10)
    dstream.pprint()
    # -------------------------------------------
    # Time: 2019-08-14 19:23:30
    # -------------------------------------------
    # hello
    ssc.start()
    ssc.awaitTermination()
            
          

reduceByKeyAndWindow是對(K,V)窗口數(shù)據相同的K執(zhí)行對應的fun，實現(xiàn)代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

def fun(x,y):
    return x+y

if __name__ == '__main__':
    sc = SparkContext(appName="windowStream", master="local[*]")
    # 第二個參數(shù)指統(tǒng)計多長時間的數(shù)據
    ssc = StreamingContext(sc, 5)
    ssc.checkpoint("/tmp/window")
    lines = ssc.socketTextStream("localhost", 9999)
    # 第一個參數(shù)執(zhí)行的功能函數(shù)fun, 第二個參數(shù)是窗口長度，這里是60秒， 第三個參數(shù)是滑動間隔，這里是10秒，
    # 第四個參數(shù)設定并行度
    dstream = lines.map(lambda x:(x,1)).reduceByKeyAndWindow(fun, 60, 10)
    dstream.pprint()
    # -------------------------------------------
    # Time: 2019-08-14 19:23:30
    # -------------------------------------------
    # ('hello', 2)
    ssc.start()
    ssc.awaitTermination()
            
          

countByValueAndWindow是對窗口數(shù)據進行單詞統(tǒng)計，實現(xiàn)代碼如下

            
              from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == '__main__':
    sc = SparkContext(appName="windowStream", master="local[*]")
    # 第二個參數(shù)指統(tǒng)計多長時間的數(shù)據
    ssc = StreamingContext(sc, 5)
    ssc.checkpoint("/tmp/window")
    lines = ssc.socketTextStream("localhost", 9999)
    # 第一個參數(shù)是窗口長度，這里是60秒， 第二個參數(shù)是滑動間隔，這里是10秒， 第三個參數(shù)任務并行度
    dstream = lines.countByValueAndWindow(60, 10)
    dstream.pprint()
    # -------------------------------------------
    # Time: 2019-08-14 19:23:30
    # -------------------------------------------
    # ('hello', 3)
    # ('world', 1)
    ssc.start()
    ssc.awaitTermination()
            
          

以上就是所有窗口函數(shù)的使用

?

Spark學習目錄：

• Spark學習實例1(Python)：單詞統(tǒng)計 Word Count
• Spark學習實例2(Python)：加載數(shù)據源Load Data Source
• Spark學習實例3(Python)：保存數(shù)據Save Data
• Spark學習實例4(Python)：RDD轉換 Transformations
• Spark學習實例5(Python)：RDD執(zhí)行 Actions
• Spark學習實例6(Python)：共享變量Shared Variables
• Spark學習實例7(Python)：RDD、DataFrame、DataSet相互轉換
• Spark學習實例8(Python)：輸入源實時處理 Input Sources Streaming
• Spark學習實例9(Python)：窗口操作 Window Operations

?

?