通常使用Spark的流式框架如Spark Streaming，做無狀態的流式計算是非常方便的，僅需處理每個批次時間間隔內的資料即可，不需要關注之前的資料，這是建立在業務需求對批次之間的資料沒有聯絡的基礎之上的。

但如果我們要跨批次做一些資料統計，比如batch是3秒，但要統計每1分鐘的使用者行為，那麼就要在整個流式鏈條中維護一個狀態來儲存近1分鐘的使用者行為。

那麼如果維護這樣一個狀態呢？一般情況下，主要通過以下幾種方式：

1. spark內建運算元：updateStateByKey、mapWithState

2. 第三方儲存系統維護狀態：如redis、alluxio、HBase這裡主要以spark內建運算元：updateStateByKey、mapWithState為例，通過一些示例程式碼（不涉及offset管理），來看看如何進行狀態維護。

updateStateByKey

分析相關原始碼發現，這個運算元的核心思想就是將之前有狀態的RDD和當前的RDD做一次cogroup，得到一個新的狀態的RDD，具有如下特點：

1. 可以設定初始狀態

2. key超時刪除。用updatefunc返回None值。updateFunc不管是否有已儲存狀態key的新資料到來，都會被已存在狀態的key呼叫，新增的key也會呼叫3. 不適合大資料量狀態儲存，尤其是key的維度比較高、value狀態比較大的

/**
* @author：微信公眾號：大資料學習與分享
*/
object StateOperator {

  private val brokers = "kafka-1:9092,kafka-2:9092,kafka-3:9092"
  private val topics = "test"
  private val groupId = "test"
  private val batchTime = "10"
  private val mapwithstateCKDir = "/mapwithstate"
  private val updateStateByKeyCKDir = "/mapwithstate"

  def main(args: Array[String]): Unit = {
    
    val ssc = StreamingContext.getOrCreate(updateStateByKeyCKDir, () => createContext(brokers, topics, groupId, batchTime, updateStateByKeyCKDir))

    ssc.start()
    ssc.awaitTermination()
  }

  def createContext(brokers: String, topics: String,
                    groupId: String, batchTime: String,
                    checkpointDirectory: String): StreamingContext = {

    val conf = new SparkConf().setAppName("testState").setMaster("local[*]")
      .set("spark.streaming.kafka.maxRatePerPartition", "5")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(conf, Seconds(batchTime.toInt))

    val topicsSet = topics.split(",").toSet
    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,
      "group.id" -> groupId,
      "auto.offset.reset" -> "smallest")

    val streams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
      .map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)

    ssc.checkpoint("/redis/updateStateByKey")

    val initialRDD = ssc.sparkContext.parallelize(List(("word", 0)))

    //updateStateByKey 底層核心是對preStateRDD（之前資料狀態的RDD）和當前批次的RDD進行cogroup
    val stateStreams = streams.updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true, initialRDD)

    stateStreams.checkpoint(Duration(5))

    stateStreams.foreachRDD { rdd =>
      val res = rdd.map { case (word, count) => (count, word) }.sortByKey(false).take(10).map { case (v, k) => (k, v) }
      res.foreach(println)
    }

    ssc.checkpoint(checkpointDirectory)
    ssc
  }

  //無論當前批次RDD有多少key(比如preStateRDD有而當前批次沒有)都需要對所有的資料進行cogroup並呼叫一次定義的updateFunc函式
  val updateFunc = (iterator: Iterator[(String, Seq[Int], Option[Int])]) => {
    iterator.flatMap(t => Some(t._2.sum + t._3.getOrElse(0)).map(v => (t._1, v)))
  }

}

通過updateStateByKey獲得的是整個狀態的資料，而且每次狀態更新時都要將當前批次過來的資料與之前儲存的狀態進行cogroup操作，並且對所有資料都呼叫自定義的函式進行一次計算。

隨著時間推移，資料量不斷增長，需要維護的狀態越來越大，會非常影響效能。如果不能在當前批次將資料處理完成，很容易造成資料堆積，影響程式穩定執行甚至宕掉，這就引出了mapWithState。

mapWithState

支援輸出全量的狀態和更新的狀態，還支援對狀態超時管理，使用者可以根據業務需求選擇需要的輸出，效能優於於updateStateByKey。

def main(args: Array[String]): Unit = {
    //單詞統計
    val ssc = StreamingContext.getOrCreate(mapwithstateCKDir,
     () => createContext(brokers, topics, groupId, batchTime, mapwithstateCKDir))

    ssc.start()
    ssc.awaitTermination()
}

def createContext(brokers: String, topics: String,
                 groupId: String, batchTime: String,
                 checkpointDirectory: String): StreamingContext = {

 val conf = new SparkConf().setAppName("testState").setMaster("local[*]")
   .set("spark.streaming.kafka.maxRatePerPartition", "5")
   .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
 val ssc = new StreamingContext(conf, Seconds(batchTime.toInt))

 val topicsSet = topics.split(",").toSet
 val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,
   "group.id" -> groupId,
   "auto.offset.reset" -> "smallest")

 val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
      .map(_._2).flatMap(_.split(" ")).map((_, 1L)).reduceByKey(_ + _)

   val stateStreams = messages.mapWithState(StateSpec.function(mapFunc).timeout(Seconds(60))).stateSnapshots()
   //.checkpoint(Duration(5))

   stateStreams.foreachRDD { (rdd, time) =>
     println("========do something")
   }

   ssc.checkpoint(checkpointDirectory)
   ssc
 }

 //key為word，value為當前批次值，state為本批次之前的狀態值
 val mapFunc = (key: String, value: Option[Long], state: State[Long]) => {
   //檢測是否過期
   if (state.isTimingOut()) {
     println(s"$key is timing out")
   } else {
     val sum = state.getOption().getOrElse(0L) + value.getOrElse(0L)
     val output = (key, sum)
     //更新狀態
     state.update(sum)
     output
   }
 }
 
 val mapFunction = (time: Time, word: String, count: Option[Int], state: State[Int]) => {
   val sum = count.getOrElse(0) + state.getOption().getOrElse(0)
   val output = (word, sum)
   state.update(sum)
   Option(output)
 }

雖然mapWithState相對於updateStateByKey效能更優，但仍然不適合大資料量的狀態維護，此時就需要借用第三方儲存來進行狀態的維護了，redis、alluxio、HBase是常用的選擇。

redis比較適合維護key具有超時處理機制的場景使用；alluxio的吞吐量更高，適合於資料量更大時的場景處理。

具體採用哪種方式，要結合實際的業務場景、資料量、效能等多方面的考量。

關注微信公眾號：大資料學習與分享，獲取更對技術