必会SparkStreaming的窗口操作及实战

Posted 2021-04-03 勾叔谈大数据

大家好，我是勾叔。今天和大家聊一下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()
}
}
 
   
   
 

大家如果想进行更深入的了解和学习，请关注勾叔谈大数据参与更多互动。

推荐阅读：