利用mapWithState實現按照首字母統計的有狀態的wordCount

最近在做sparkstreaming整合kafka的時候遇到了一個問題：

可以抽象成這樣一個問題：有狀態的wordCount，且按照word的第一個字母為key，但是要求輸出的格式為(word,1)這樣的形式

舉例來說：

例如第一批數據為: hello how when hello

則要求輸出為：(hello,1) (how,2) (when,1) (hello,3)

第二批數據為: hello how when what hi

則要求輸出為: (hello,4) (how,5) (when,2) (what,3) (hi,6)

首先了解一下mapWithState的常規用法:

ref: https://www.jianshu.com/p/a54b142067e5

http://sharkdtu.com/posts/spark-streaming-state.html

稍微總結一下mapWithState的幾個tips:

1. mapWithState是1.6版本之后推出的
2. 必須設置checkpoint來儲存歷史數據
3. mapWithState和updateStateByKey的區別 : 他們類似,都是有狀態DStream操作, 區別在於,updateStateByKey是輸出增量數據,隨着時間的增加, 輸出的數據越來越多,這樣會影響計算的效率, 對CPU和內存壓力較大.而mapWithState則輸出本批次數據,但是也含有狀態更新.
4. checkpoint的數據會分散存儲在不同的分區中, 在進行狀態更新時, 首先會對當前 key 做 hash , 再到對應的分區中去更新狀態 , 這種方式大大提高了效率.

解決問題的思路:

State中保存狀態為(String,Int) 元組類型, 其中String為word的全量, 而Int為word的計數.

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.MapWithStateDStream
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}

object MapWithStateApp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("MapWithStateApp")
    val ssc = new StreamingContext(conf,Seconds(5))
    ssc.checkpoint("C:\\Users\\hylz\\Desktop\\checkpoint")
    val lines = ssc.socketTextStream("192.168.100.11",8888)
    val words = lines.flatMap(_.split(" "))

    def mappingFunc(key: String, value: Option[(String, Int)], state: State[(String, Int)]): (String, Int) = {
      val cnt: Int = value.getOrElse((null, 0))._2 + state.getOption.getOrElse((null, 0))._2
      val allField: String = value.getOrElse((null, 0))._1
      state.update((allField, cnt))
      (allField, cnt)
    }

    val cnt: MapWithStateDStream[String, (String, Int), (String, Int), (String, Int)] = words.map(x => (x.substring(0, 1), (x, 1))).mapWithState(StateSpec.function(mappingFunc _))

    cnt.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

測試結果如下

input： hello how when hello

input： hello how when what hi