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

文章目錄

關於mapWithState

  須要本身寫一個匿名函數func來實現本身想要的功能。若是有初始化的值得須要，可使用initialState(RDD)來初始化key的值。 另外，還能夠指定timeout函數，該函數的做用是，若是一個key超過timeout設定的時間沒有更新值，那麼這個key將會失效。這個控制須要在func中實現，必須使用state.isTimingOut()來判斷失效的key值。若是在失效時間以後，這個key又有新的值了，則會從新計算。若是沒有使用isTimingOut，則會報錯。

注意事項

  下面程序是使用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端運行編寫好的程序
  服務器發送數據

  控制檯顯示結果

結論

  mapWithState它會按照時間線在每個批次間隔返回以前的發生改變的或者新的key的狀態，不發生變化的不返回。同時mapWithState能夠不用設置checkpoint，返回的數據量少，性能和效率都比mapWithState好。