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

Window Operations（窗口操作）可以设置窗口大小和滑动窗口间隔来动态的获取当前Streaming的状态。基于窗口的操作会在一个比 StreamingContext 的 batchDuration（批次间隔）更长的时间范围内，通过整合多个批次的结果，计算出整个窗口的结果。

下面，通过一张图来描述SparkStreaming的窗口操作，如图所示。

基于窗口的操作需要两个参数，如下：

• 窗口长度(windowDuration)，控制每次计算最近的多少个批次的数据；

• 滑动间隔(slideDuration)，用来控制对新的 DStream 进行计算的间隔。

两者都必须是 StreamingContext 中批次间隔(batchDuration)的整数倍。

使用窗口操作，即使用窗口操作进行实战。

每秒发送1个数字

package cn.lagou.streaming
import java.io.PrintWriter
import java.net.{ServerSocket, Socket}
object SocketLikeNCWithWindow {
    def main(args: Array[String]): Unit = {
        val port = 1521
        val ss = new ServerSocket(port)
        val socket: Socket = ss.accept()
        println("connect to host : " + socket.getInetAddress)
        var i = 0
        // 每秒发送1个数
        while(true) {
            i += 1
            val out = new PrintWriter(socket.getOutputStream)
            out.println(i)
            out.flush()
            Thread.sleep(1000)
        }
    }
}

案例一

观察窗口的数据；观察 batchDuration、windowDuration、slideDuration 三者之间的关系；使用窗口相关的操作，具体代码演示如下：

package cn.lagou.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object WindowDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]")
                                  .setAppName(this.getClass.getCanonicalName)
        // 每 5s 生成一个RDD（mini-batch）
        val ssc = new StreamingContext(conf, Seconds(5))
        ssc.sparkContext.setLogLevel("error")
        val lines: ReceiverInputDStream[String] =
        ssc.socketTextStream("localhost", 1521)
        lines.foreachRDD{ (rdd, time) =>println(s"rdd = ${rdd.id}; time = $time")
            rdd.foreach(value => println(value))
        }
        val res1: DStream[String] =lines.reduceByWindow(_ + " " + _,Seconds(20), Seconds(10))
        res1.print()
        val res2: DStream[String] = lines.window(Seconds(20),Seconds(10))
        res2.print()
        // 求窗口元素的和
        val res3:DStream[Int]=lines.map(_.toInt).reduceByWindow(_+_,Seconds(20), Seconds(10))
        res3.print()
        // 求窗口元素的和
        val res4 = res2.map(_.toInt).reduce(_+_)
        res4.print()
        ssc.start()
        ssc.awaitTermination()
    }
}

案例二

热点搜索词实时统计。每隔 10 秒，统计最近20秒的词出现的次数，具体代码演示如下：

package cn.lagou.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object HotWordStats {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[2]")
                              .setAppName(this.getClass.getCanonicalName)
        val ssc = new StreamingContext(conf, Seconds(2))
        ssc.sparkContext.setLogLevel("ERROR")
        //设置检查点，检查点具有容错机制。生产环境中应设置到HDFS
        ssc.checkpoint("data/checkpoint/")
        val lines: ReceiverInputDStream[String] =ssc.socketTextStream("localhost", 9999)
        val words: DStream[String] = lines.flatMap(_.split("\\s+"))
        val pairs: DStream[(String, Int)] = words.map(x => (x, 1))
        // 通过reduceByKeyAndWindow算子, 每隔10秒统计最近20秒的词出现的次数
        // 后 3个参数：窗口时间长度、滑动窗口时间、分区
        val wordCounts1: DStream[(String, Int)] =pairs.reduceByKeyAndWindow(
                                    (a: Int, b: Int) => a + b,Seconds(20),Seconds(10), 2)
        wordCounts1.print
        // 这里需要checkpoint的支持
        val wordCounts2: DStream[(String, Int)] = pairs.reduceByKeyAndWindow(_ + _,_ - _,
                                                  Seconds(20),Seconds(10), 2)
        wordCounts2.print
        ssc.start()
        ssc.awaitTermination()
    }
}