0


DStream窗口操作

在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节点每隔一秒输入字母,运行结果如下:


本文转载自: https://blog.csdn.net/m0_59839948/article/details/125341121
版权归原作者 鄙人阿彬 所有, 如有侵权,请联系我们删除。

“DStream窗口操作”的评论:

还没有评论