Spark Streaming狀態管理函數（二）——updateStateByKey的使用（scala版）

 

關於updateStateByKey

        1.重點：首先會以DStream中的數據進行按key作reduce操做，而後再對各個批次的數據進行累加 。

        2.updateStateBykey要求必需要設置checkpoint點。

        3.updateStateByKey 方法中 updateFunc就要傳入的參數，。Seq[V]表示當前key對應的全部值，Option[S] 是當前key的歷史狀態，返回的是新的封裝的數據。

注意事項

        下面程序是使用idea編寫的，使用的是scala語言，在程序中master(「local[2]」)設置爲本地模式（[]中的數指定的是線程數，不能少於2，不然看不到結果。主要是由於spark須要啓動一個線程receiver來循環接收數據，一個Executor來接收數據，若是少於2線程不夠將不能打印出結果。），在window上運行的。使用的spark版本是2.3.0，在2.x之後的版本，基本採用SparkSession來進行操做。同時，想要運行程序你的服務器上還必需要安裝netcat這個軟件，使用yum install nc進行安裝（注意安全配置好yum源，DNS才能下載安裝），使用命令nc -lk 6666開啓服務發送數據。最後在運行程序前還須要導入spark、scala相應的依賴包。

示例代碼

package spark2x
 
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
/**
  * 類名  UpdateStateByKeyDemo
  * 做者   彭三青
  * 建立時間  2018-12-01 9:35
  * 版本  1.0
  * 描述: $
  */

object UpdateStateByKeyDemo {
  def main(args: Array[String]): Unit = {
    /** 第一步：配置SparkConf：
      * 1，至少2條線程：由於Spark Streaming應用程序在運行的時候，至少有一條
      * 線程Receiver用於不斷的循環接收數據，還有一條線程是Executor用於處理接受的數據（少於兩條
      * 就沒有線程用於處理數據，窗口不會顯示數據。而且隨着時間的推移，內存和磁盤因爲負擔太重而崩潰）；
      * 2，對於集羣而言，根據已有經驗，大概5個左右的Core是性能最佳（通常分配爲奇數個Core）
      */
     val spark = SparkSession.builder()
      .master("local[2]")
      .appName("UpdateStateByKeyDemo")
      .getOrCreate()
    val conf: SparkContext = spark.sparkContext
    /**
      * 第二步：建立SparkStreamingContext：
      * 1，這個是SparkStreaming應用程序全部功能的起始點和程序調度的核心
      * SparkStreamingContext的構建能夠基於SparkConf參數，也可基於持久化的SparkStreamingContext的內容
      * 來恢復過來（典型的場景是Driver崩潰後從新啓動，因爲Spark Streaming具備連續7*24小時不間斷運行的特徵，
      * 全部須要在Driver從新啓動後繼續上衣系的狀態，此時的狀態恢復須要基於曾經的Checkpoint）；
      * 2，在一個Spark Streaming應用程序中能夠建立若干個SparkStreamingContext對象，使用下一個SparkStreamingContext
      * 以前須要把前面正在運行的SparkStreamingContext對象關閉掉，由此，咱們得到一個重大的啓發SparkStreaming框架也只是
      * Spark Core上的一個應用程序而已，只不過Spark Streaming框架箱運行的話須要Spark工程師寫業務邏輯處理代碼；
      */
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //報錯解決辦法作checkpoint,開啓checkpoint機制，把checkpoint中的數據放在這裏設置的目錄中，生產環境下通常放在HDFS中
    ssc.checkpoint("hdfs://SC01:8020/user/tmp/cp-20181201")

    /**
      * 第三步：建立Spark Streaming輸入數據來源input Stream：
      * 1，數據輸入來源能夠基於File、HDFS、Flume、Kafka、Socket等
      * 2, 在這裏咱們指定數據來源於網絡Socket端口，Spark Streaming鏈接上該端口並在運行的時候一直監聽該端口
      * 的數據（固然該端口服務首先必須存在）,而且在後續會根據業務須要不斷的有數據產生(固然對於Spark Streaming
      * 應用程序的運行而言，有無數據其處理流程都是同樣的)；
      * 3,若是常常在每間隔5秒鐘沒有數據的話不斷的啓動空的Job實際上是會形成調度資源的浪費，由於並無數據須要發生計算，因此
      * 實例的企業級生成環境的代碼在具體提交Job前會判斷是否有數據，若是沒有的話就再也不提交Job；
      */
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("SC01", 6666)

    /**
      * 第四步：接下來就像對於RDD編程同樣基於DStream進行編程！！！緣由是DStream是RDD產生的模板（或者說類），在Spark Streaming具體
      * 發生計算前，其實質是把每一個Batch的DStream的操做翻譯成爲對RDD的操做！！！
      * 對初始的DStream進行Transformation級別的處理，例如map、filter等高階函數等的編程，來進行具體的數據計算
      * 進行單詞拆分
      */
    val words: DStream[String] = line.flatMap(_.split(" "))

    /**
      * 對初始的DStream進行Transformation級別的處理，例如map、filter等高階函數等的編程，來進行具體的數據計算
      * 單詞分組計數實，word => (word, 1) Word ->(word, 1) day -> day(day, 1)
      */
    val pairs: DStream[(String, Int)] = words.map((_, 1))

    /**
      * 經過updateStateByKey來以Batch Interval爲單位來對歷史狀態進行更新，
      * 這是功能上的一個很是大的改進，不然的話須要完成一樣的目的，就可能須要把數據保存在Redis、
      * Tagyon或者HDFS或者HBase或者數據庫中來不斷的完成一樣一個key的State更新，若是你對性能有極爲苛刻的要求，
      * 且數據量特別大的話，能夠考慮把數據放在分佈式的Redis或者Tachyon內存文件系統中；
      * Spark2.X後mapWithState應該很是穩定了。
      */
    val wordCount: DStream[(String, Int)] = pairs.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
      var newValue = state.getOrElse(0)
      for (value <- values) {
        newValue += value
      }
      Option(newValue)
    })

    /**
      * 此處的print並不會直接出發Job的執行，由於如今的一切都是在Spark Streaming框架的控制之下的，對於Spark Streaming
      * 是否觸發真正的Job運行是基於設置的Duration時間間隔的
      * 須要注意的是Spark Streaming應用程序要想執行具體的Job，對Dtream就必須有output Stream操做，
      * output Stream有不少類型的函數觸發，類print、saveAsTextFile、saveAsHadoopFiles等，最爲重要的一個
      * 方法是foraeachRDD,由於Spark Streaming處理的結果通常都會放在Redis、DB、DashBoard等上面，foreachRDD
      * 主要就是用用來完成這些功能的，並且能夠隨意的自定義具體數據到底放在哪裏！！！
      */
    wordCount.print()

    /**
  * Spark Streaming執行引擎也就是Driver開始運行，Driver啓動的時候是位於一條新的線程中的，固然其內部有消息循環體，用於
  * 接受應用程序自己或者Executor中的消息；
      */
    // 開始提交任務
    ssc.start()
    // 線程等待，等待處理下一批次任務
    ssc.awaitTermination()
  }

    /** Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)]
      * 在調用updateStateByKey中，須要傳入一個用於計算曆史批次和當前批次數據的函數
      * 該函數中有幾個類型：String, Seq[Int], Option[Int])]
      * String表明元組中每個單詞，也就是key
      * Seq[Int]表明當前批次相同key對應的value，好比Seq(1,1,1,1)
      * Option[Int]表明上一批次中相同key對應的累加的結果，有可能有值，有可能沒有值。
      * 此時，獲取歷史批次的數據時，最好用getOrElse方法
      */
  val func = (it: Iterator[(String, Seq[Int], Option[Int])]) => {
    it.map(tup => {
      (tup._1, tup._2.sum + tup._3.getOrElse(0))
    })
  }
}

運行

        服務器運行nc

        在idea端運行編寫好的程序
   服務器發送數據

        控制檯顯示結果

結論

 

        updateStateByKey它會按照時間線在每個批次間隔返回以前的key的狀態，它會對已存在的key進行state的狀和每一個新出現的key執行相同的更新函數操做。同時updateStateByKey，要求必需要設置checkpoint，對數據進行緩存。