視窗的一些簡單操作

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object windowOpObj {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("windowOpObj").setMaster("local[*]")
    //建立
   val ssc= new StreamingContext(conf,Seconds(5))

    //儲存不同（之前）批次的RDD資料
    ssc.checkpoint("D:\\資料\\spark")
    //獲取資料
    val datream = ssc.socketTextStream("192.168.88.130",8888)
    //呼叫視窗函式來聚合多個批次的資料,現在的操作基於每一個批次
    val tupls = datream.flatMap(_.split(" ")).map((_,1))
    //設定視窗長度以及滑動時間間隔
    //處理多個批次,2引數是視窗長度，第3個引數是視窗的滑動時間，可以計算指定範圍內的批次資料
   val res= tupls.reduceByKeyAndWindow((x:Int,y:Int)=>(x+y),Seconds(15),Seconds(10))
    res.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

/*
一下是視窗滑動的時間
Time: 1541809190000 ms

Time: 1541809200000 ms
-------------------------------------------

Time: 1541809210000 ms
-------------------------------------------

Time: 1541809220000 ms
-------------------------------------------

Time: 1541809230000 ms

*/