068 mapWithState函數的講解

1.問題

　　主要是updateStateByKey的問題

　　有的值不需要變化的時候，還會再打印出來。

　　每個批次的數據都會出現，如果向redis保存更新的時候，會把不需要變化的值也更新，這個不是我們需要的，我們只需要更新有變化的那部分值。

　　

2.mapWithState

　　有一個注解，說明是實驗性質的。

　　

 

3.程序

package com.stream.it
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object MapWithState {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("StreamingMapWithState")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    val ssc = new StreamingContext(sc, Seconds(1))
    // 當調用updateStateByKey函數API的時候，必須給定checkpoint dir
    // 路徑對應的文件夾不能存在
    ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/chkdir45254")

    /**
      *
      * @param key    DStream的key數據類型
      * @param values DStream的value數據類型
      * @param state  是StreamingContext中之前該key的狀態值
      * @return
      */
    def mappingFunction(key: String, values: Option[Int], state: State[Long]): (String, Long) = {
      // 獲取之前狀態的值
      val preStateValue = state.getOption().getOrElse(0L)
      // 計算出當前值
      val currentStateValue = preStateValue + values.getOrElse(0)

      // 更新狀態值
      state.update(currentStateValue)

      // 返回結果
      (key, currentStateValue)
    }
    val spec = StateSpec.function[String, Int, Long, (String, Long)](mappingFunction _)

    val kafkaParams = Map(
      "group.id" -> "streaming-kafka-001231",
      "zookeeper.connect" -> "linux-hadoop01.ibeifeng.com:2181/kafka",
      "auto.offset.reset" -> "smallest"
    )
    val topics = Map("beifeng" -> 4) // topics中value是讀取數據的線程數量，所以必須大於等於1
    val dstream = KafkaUtils.createStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](
      ssc, // 給定SparkStreaming上下文
      kafkaParams, // 給定連接kafka的參數信息 ===> 通過Kafka HighLevelConsumerAPI連接
      topics, // 給定讀取對應topic的名稱以及讀取數據的線程數量
      StorageLevel.MEMORY_AND_DISK_2 // 指定數據接收器接收到kafka的數據后保存的存儲級別
    ).map(_._2)

    val resultWordCount: DStream[(String, Long)] = dstream
      .filter(line => line.nonEmpty)
      .flatMap(line => line.split(" ").map((_, 1)))
      .reduceByKey(_ + _)
      .mapWithState(spec)

    resultWordCount.print() // 這個也是打印數據

    // 啟動開始處理
    ssc.start()
    ssc.awaitTermination() // 等等結束，監控一個線程的中斷操作
  }
}

 

4.效果

　　

　　在控制台上再寫入一個hadoop：

　　　　說明了，在新寫入的時候，才會出現，但是以前的數據還在。

　　

 

5.說明

　　因為存在checkpoint，在重新后，以前的數據還在，新加入數據后，會在原有的基礎上進行更新，上面的第二幅圖就是這樣產生的。