Window Operations（窗口操作）可以设置窗口大小和滑动窗口间隔来动态的获取当前Streaming的状态。基于窗口的操作会在一个比 StreamingContext 的 batchDuration（批次间隔）更长的时间范围内，通过整合多个批次的结果，计算出整个窗口的结果。
下面，通过一张图来描述SparkStreaming的窗口操作，如图所示。
基于窗口的操作需要两个参数，如下：
窗口长度(windowDuration)，控制每次计算最近的多少个批次的数据；
滑动间隔(slideDuration)，用来控制对新的 DStream 进行计算的间隔。
两者都必须是 StreamingContext 中批次间隔(batchDuration)的整数倍。
接下来，勾叔带大家来使用窗口操作，即使用窗口操作进行实战。
每秒发送1个数字
```
import java.io.PrintWriter
import java.net.{ServerSocket,Socket}
objectSocketLikeNCWithWindow{
defmain(args:Array[String]):Unit={
valport=1521
valss=newServerSocket(port)
valsocket:Socket=ss.accept()
println("connect to host : "+socket.getInetAddress)
vari=0
// 每秒发送1个数
while(true) {
i+=1
valout=newPrintWriter(socket.getOutputStream)
out.println(i)
out.flush()
Thread.sleep(1000)
}
}
}
案例一
观察窗口的数据；观察 batchDuration、windowDuration、slideDuration 三者之间的关系；使用窗口相关的操作，具体代码演示如下：
packagecn.lagou.streaming
importorg.apache.spark.SparkConf
importorg.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}
importorg.apache.spark.streaming.{Seconds,StreamingContext}
objectWindowDemo{
defmain(args:Array[String]):Unit={
valconf=newSparkConf().setMaster("local[*]")
.setAppName(this.getClass.getCanonicalName)
// 每 5s 生成一个RDD（mini-batch）
valssc=newStreamingContext(conf,Seconds(5))
ssc.sparkContext.setLogLevel("error")
vallines:ReceiverInputDStream[String]=
ssc.socketTextStream("localhost",1521)
lines.foreachRDD{ (rdd,time)=>println(s"rdd = ${rdd.id}; time = $time")
rdd.foreach(value=>println(value))
}
valres1:DStream[String]=lines.reduceByWindow(_+" "+_,Seconds(20),Seconds(10))
res1.print()
valres2:DStream[String]=lines.window(Seconds(20),Seconds(10))
res2.print()
// 求窗口元素的和
valres3:DStream[Int]=lines.map(_.toInt).reduceByWindow(_+_,Seconds(20),Seconds(10))
res3.print()
// 求窗口元素的和
valres4=res2.map(_.toInt).reduce(_+_)
res4.print()
ssc.start()
ssc.awaitTermination()
}
}
案例二
热点搜索词实时统计。每隔 10 秒，统计最近20秒的词出现的次数，具体代码演示如下：
packagecn.lagou.streaming
importorg.apache.spark.SparkConf
importorg.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}
importorg.apache.spark.streaming.{Seconds,StreamingContext}
objectHotWordStats{
defmain(args:Array[String]):Unit={
valconf:SparkConf=newSparkConf().setMaster("local[2]")
.setAppName(this.getClass.getCanonicalName)
valssc=newStreamingContext(conf,Seconds(2))
ssc.sparkContext.setLogLevel("ERROR")
//设置检查点，检查点具有容错机制。生产环境中应设置到HDFS
ssc.checkpoint("data/checkpoint/")
vallines:ReceiverInputDStream[String]=ssc.socketTextStream("localhost",9999)
valwords:DStream[String]=lines.flatMap(_.split("\\s+"))
valpairs:DStream[(String,Int)]=words.map(x=>(x,1))
// 通过reduceByKeyAndWindow算子, 每隔10秒统计最近20秒的词出现的次数
// 后 3个参数：窗口时间长度、滑动窗口时间、分区
valwordCounts1:DStream[(String,Int)]=pairs.reduceByKeyAndWindow(
(a:Int,b:Int)=>a+b,Seconds(20),Seconds(10),2)
wordCounts1.print
// 这里需要checkpoint的支持
valwordCounts2:DStream[(String,Int)]=pairs.reduceByKeyAndWindow(_+_,_-_,
Seconds(20),Seconds(10),2)
wordCounts2.print
ssc.start()
ssc.awaitTermination()
}
}
```
