博客

1、Window

 窗口的范围应该是采集周期的整数倍
窗口是可以滑动的，但是默认情况下，一个采集是按照周期滑动的
这样的话，可能造成出现重复数据的计算，为了避免这种情况可以改变滑动的幅度（步长）
关于窗口的两个参数                                      窗口的时长     滑动的步长
窗口的时长：计算内容的时间范围
滑动的步长：隔多久触发一次计算

package com.zch.spark.streaming
​
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
​
/**
 * Author: zhaoHui
 * Date: 2022/01/03
 * Time: 19:03
 * Description: 有状态操作
 */
object sparkStreaming05_State_Window {
  def main(args: Array[String]): Unit = {
​
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("window")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
​
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val wordToOne: DStream[(String, Int)] = lines.map((_, 1))
​
    // 窗口的范围应该是采集周期的整数倍
    // 窗口是可以滑动的，但是默认情况下，一个采集是按照周期滑动的
    // 这样的话，可能造成出现重复数据的计算，为了避免这种情况可以改变滑动的幅度（步长）
    // 关于窗口的两个参数                                      窗口的时长     滑动的步长
    // 窗口的时长：计算内容的时间范围
    // 滑动的步长：隔多久触发一次计算
    val windowDS: DStream[(String, Int)] = wordToOne.window(Seconds(6),Seconds(6))
​
    val wordToCount: DStream[(String, Int)] = windowDS.reduceByKey(_ + _)
​
    wordToCount.print()
​
    ssc.start()
    ssc.awaitTermination()
  }
​
}

2、reduceByKeyAndWindow

package com.zch.spark.streaming

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

/**
 * Author: zhaoHui
 * Date: 2022/01/03
 * Time: 19:03
 * Description: 有状态操作
 */
object sparkStreaming05_State_reduceByKeyWindow {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("reduceByKeyAndWindow")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("F:\\JAVA\\bigdata2107\\zch\\spark\\src\\main\\resources")

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val wordToOne: DStream[(String, Int)] = lines.map((_, 1))

    // reduceByKeyAndWindow:当窗口的范围较大，但是滑动幅度比较小，那么可以采用增加数据和
    // 删除数据的功能，无重复计算，提升性能
    val windowDS: DStream[(String, Int)] = wordToOne
      .reduceByKeyAndWindow(
      (x: Int, y: Int) => {
        x + y
      },
      (x: Int, y: Int) => {
        x - y
      },
      Seconds(9), Seconds(3)
    )

    windowDS.print()

    ssc.start()
    ssc.awaitTermination()
  }

}