DStream窗口操作

Posted 2022-06-20 鄙人阿彬

在SparkStreaming中，为DStream提供窗口操作，即在 DStream流上，将一个可配置的长度设置为窗口，以一个可配置的速率向前移动窗口。根据窗口操作，对窗口内的数据进行计算，每次落在窗口内的RDD数据会被聚合起来计算，生成的RDD会作为WindowDStream的一个RDD。

DStream API提供的与窗口操作相关的方法：

 

 

 

 

 1、window（）

在spark03项目的/src/main/scala/itcast目录下创建一个名为WindowTest的scala类，编写以下内容：

import org.apache.spark.streaming.dstream.DStream, ReceiverInputDStream
import org.apache.spark.streaming.Seconds, StreamingContext
import org.apache.spark.SparkConf, SparkContext

object WindowTest 
  def main(args: Array[String]): Unit = 
    //创建SparkConf对象
    val sparkConf : SparkConf = new SparkConf()
      .setAppName("WindowTest").setMaster("local[2]")

    //创建SparkContext对象
    val sc : SparkContext = new SparkContext(sparkConf)
    //设置日志级别
    sc.setLogLevel("WARN")

    //创建StreamingContext
    val ssc : StreamingContext = new StreamingContext(sc,Seconds(1))

    //连接socket服务
    val dstream: ReceiverInputDStream[String] =
      ssc.socketTextStream("192.168.196.101", 9999)
    //按空格切分每一行
    val words:DStream[String]= dstream.flatMap(_.split(""))

    //调用Window操作，需要两个参数
     val windowWords : DStream[String]= words.window(Seconds(3),Seconds(1))

    //打印输出
    windowWords.print()
    //开启流式计算
    ssc.start()
    //用于保持程序一直运行
    ssc.awaitTermination()



  

先在IDEA中的WindowTest运行代码，然后再到master节点每隔一秒输入数字，运行结果如下：

 

 

2、reduceByKeyAndWindow（）

在spark03项目的/src/main/scala/itcast目录下创建一个名为ReduceByKeyAndWindowTest的scala类，编写以下内容：

import org.apache.spark.streaming.Seconds, StreamingContext
import org.apache.spark.streaming.dstream.DStream, ReceiverInputDStream
import org.apache.spark.SparkConf, SparkContext

object ReduceByKeyAndWindowTest 
  def main(args: Array[String]): Unit = 
    //创建SparkConf对象
    val sparkConf : SparkConf = new SparkConf()
      .setAppName("ReduceByKeyAndWindowTest").setMaster("local[2]")

    //创建SparkContext对象
    val sc : SparkContext = new SparkContext(sparkConf)
    //设置日志级别
    sc.setLogLevel("WARN")

    //创建StreamingContext
    val ssc : StreamingContext = new StreamingContext(sc,Seconds(1))

    //连接socket服务
    val dstream: ReceiverInputDStream[String] =
      ssc.socketTextStream("192.168.196.101", 9999)
    //按空格切分每一行
    val wordAnOne : DStream[(String,Int)]= dstream.flatMap(_.split(" ")).map(word =>(word,1))
    //调用ReduceByKeyAndWindowTest操作
    val windowWords:DStream[(String,Int)]= wordAnOne.reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),Seconds(3),Seconds(1))

    //打印输出
    windowWords.print()
    //开启流式计算
    ssc.start()
    //用于保持程序一直运行
    ssc.awaitTermination()

  

先在IDEA中的ReduceByKeyAndWindowTest运行代码，然后再到master节点每隔一秒输入字母，运行结果如下：