updateStateByKey的使用

  關於updateStateByKey

  注意事項

  示例程式碼

  執行

  結論

關於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，對資料進行快取。

第一篇：Spark Streaming狀態管理函式（一）——updateStateByKey和mapWithState
第三篇：Spark Streaming狀態管理函式（三）——MapWithState的使用（scala版）