1、UpdateStateByKey Spark是以采集周期处理数据，可以做到秒级。因此统计的是当前周期的数据汇总，但有些场景我们需要统计累加数据，例如：当天登录人数累加，当天某个广告的点击量累计，按小时统计首次登录人数等。这里我们需要用到以前周期的数据，因此我们需要将周期内的统计结果保存起来，用于下个周期使用。其实，我们自己也可以使用Redis缓存类工具做手动保存。这里SparkStreaming提供了状态维持的方法，便于操作，简单易用。
UpdateStateByKey 原语用于记录历史记录，有时，我们需要在 DStream 中跨批次维护状态(例如流计算中累加 wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对。
案例：这里用状态维持，实现累计词频统计，数据源采用kafka数据源。
核心代码：
package cn.streaming.status.keepimport java.utilimport cn.streaming.kafkaStream.MyKafkaUtilimport org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.spark.SparkConfimport 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.windowimport java.utilimport cn.streaming.kafkaStream.MyKafkaUtilimport org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.spark.SparkConfimport 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) rec.print() ssc.start() ssc.awaitTermination() }}
启动测试：
可以看出，这里已经是滑动采集。