SparkStreaming状态操作和滑动窗口

​ Spark是以采集周期处理数据，可以做到秒级。因此统计的是当前周期的数据汇总，但有些场景我们需要统计累加数据，例如：当天登录人数累加，当天某个广告的点击量累计，按小时统计首次登录人数等。这里我们需要用到以前周期的数据，因此我们需要将周期内的统计结果保存起来，用于下个周期使用。其实，我们自己也可以使用Redis缓存类工具做手动保存。这里SparkStreaming提供了状态维持的方法，便于操作，简单易用。

1、UpdateStateByKey

​ UpdateStateByKey 原语用于记录历史记录，有时，我们需要在 DStream 中跨批次维护状态(例如流计算中累加 wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对。

案例：这里用状态维持，实现累计词频统计，数据源采用kafka数据源。

核心代码：

package cn.streaming.status.keep
import java.util
import cn.streaming.kafkaStream.MyKafkaUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object AcculumateWordCount {
  def main(args: Array[String]): Unit = {

      val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("acc_wo")

      val ssc = new StreamingContext(conf, Seconds(3))
      
      // 使用状态操作必须使用检查点  
      ssc.checkpoint("./acc")

      val topics: util.List[String] = util.Arrays.asList("acc_wo")

      val kfkstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(ssc, topics)

      val words: DStream[(String, Int)] = kfkstream.map(_.value()).flatMap(_.split(" ")).map(x => (x, 1))
      // 使用状态方法
      val rec: DStream[(String, Int)] = words.updateStateByKey(
        (seq: Seq[Int], buff: Option[Int]) => {
          // seq：表示当前数据中，某个相同key的value的集合
          // buff：表示缓冲区中，该key的value值
          val newcount = seq.sum + buff.getOrElse(0)
          Option(newcount)
        })
      // 控制台输出
      rec.print()

      ssc.start()
      ssc.awaitTermination()
  }
}

启动主程序

启动kafka生产者并发送数据

可以看出，数据被累计统计了

2、WindowOperations

滑动窗口操作。我们可以统计多个采集周期的数据总量，然后将窗口向后滑动

通过滑动窗口，可以统计单个小时区间内的数据总量。

这里还是以词频统计为案例

实例代码：

package cn.streaming.status.window
import java.util
import cn.streaming.kafkaStream.MyKafkaUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
object AcculumateWordcountWindow {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("acc_wo")
    val ssc = new StreamingContext(conf, Seconds(3))
    // 使用状态操作必须使用检查点
    ssc.checkpoint("./acc")
    val topics: util.List[String] = util.Arrays.asList("acc_wo")
    val kfkstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(ssc, topics)
    val words: DStream[(String, Int)] = kfkstream.map(_.value()).flatMap(_.split(" ")).map((_, 1))
    // 使用滑动窗口
    val rec: DStream[(String, Int)] = words.reduceByKeyAndWindow((x:Int, y:Int) => (x + y), Seconds(12), Seconds(12)
    /**
     * reduceFunc: (V, V) => V,
     * windowDuration: Duration,
     * slideDuration: Duration
     * 
     * 参数一：聚合的函数
     * 参数二：窗口的大小，一般设置采集周期的整数倍
     * 参数三：窗口滑动的大小，当窗口大小和滑动大小相同时，即采集大区间的数据
     */
    rec.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

启动测试：

可以看出，这里已经是滑动采集。