【必会】SparkStreaming的窗口操作及实战

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()

    }

}

```