Spark Streaming updateStateByKey案例實戰和內幕源碼解密

本節課程主要分二個部分：

1、Spark Streaming updateStateByKey案例實戰
2、Spark Streaming updateStateByKey源碼解密

第一部分：

updateStateByKey的主要功能是隨着時間的流逝，在Spark Streaming中能夠爲每個能夠經過CheckPoint來維護一份state狀態，經過更新函數對該key的狀態不斷更新；對每個新批次的數據（batch）而言，Spark Streaming經過使用updateStateByKey爲已經存在的key進行state的狀態更新（對每一個新出現的key，會一樣執行state的更新函數操做）；可是若是經過更新函數對state更新後返回none的話，此時刻key對應的state狀態被刪除掉，須要特別說明的是state能夠是任意類型的數據結構，這就爲咱們的計算帶來無限的想象空間；

很是重要：

若是要不斷的更新每一個key的state，就必定會涉及到狀態的保存和容錯，這個時候就須要開啓checkpoint機制和功能，須要說明的是checkpoint的數據能夠保存一些存儲在文件系統上的內容，例如：程序未處理的但已經擁有狀態的數據。

補充說明：

關於流式處理對歷史狀態進行保存和更新具備重大實用意義，例如進行廣告（投放廣告和運營廣告效果評估的價值意義，熱點隨時追蹤、熱力圖）

案例實戰源碼：

1.編寫源碼：

ackage org.apache.spark.examples.streaming;

import java.util.Arrays;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.streaming.Durations;

import org.apache.spark.streaming.api.java.JavaDStream;

import org.apache.spark.streaming.api.java.JavaPairDStream;

import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;

import org.apache.spark.streaming.api.java.JavaStreamingContext;

import com.google.common.base.Optional;

import scala.Tuple2;

 

public class UpdateStateByKeyDemo {

public static void main(String[] args) {

/*

* 第一步：配置SparkConf：

* 1，至少2條線程：由於Spark Streaming應用程序在運行的時候，至少有一條

* 線程用於不斷的循環接收數據，而且至少有一條線程用於處理接受的數據（不然的話沒法

* 有線程用於處理數據，隨着時間的推移，內存和磁盤都會不堪重負）；

* 2，對於集羣而言，每一個Executor通常確定不止一個Thread，那對於處理Spark Streaming的

* 應用程序而言，每一個Executor通常分配多少Core比較合適？根據咱們過去的經驗，5個左右的

* Core是最佳的（一個段子分配爲奇數個Core表現最佳，例如3個、5個、7個Core等）；

*/

SparkConf conf = new SparkConf().setMaster("local[2]").

setAppName("UpdateStateByKeyDemo");

/*

* 第二步：建立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工程師寫業務邏輯處理代碼；

*/

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

//報錯解決辦法作checkpoint,開啓checkpoint機制，把checkpoint中的數據放在這裏設置的目錄中，

//生產環境下通常放在HDFS中

jsc.checkpoint("/usr/local/tmp/checkpoint");

/*

* 第三步：建立Spark Streaming輸入數據來源input Stream：

* 1，數據輸入來源能夠基於File、HDFS、Flume、Kafka、Socket等

* 2, 在這裏咱們指定數據來源於網絡Socket端口，Spark Streaming鏈接上該端口並在運行的時候一直監聽該端口

* 的數據（固然該端口服務首先必須存在）,而且在後續會根據業務須要不斷的有數據產生(固然對於Spark Streaming

* 應用程序的運行而言，有無數據其處理流程都是同樣的)； 

* 3,若是常常在每間隔5秒鐘沒有數據的話不斷的啓動空的Job實際上是會形成調度資源的浪費，由於並無數據須要發生計算，因此

* 實例的企業級生成環境的代碼在具體提交Job前會判斷是否有數據，若是沒有的話就再也不提交Job；

*/

JavaReceiverInputDStream lines = jsc.socketTextStream("hadoop100", 9999);

/*

* 第四步：接下來就像對於RDD編程同樣基於DStream進行編程！！！緣由是DStream是RDD產生的模板（或者說類），在Spark Streaming具體

* 發生計算前，其實質是把每一個Batch的DStream的操做翻譯成爲對RDD的操做！！！

*對初始的DStream進行Transformation級別的處理，例如map、filter等高階函數等的編程，來進行具體的數據計算

    * 第4.1步：講每一行的字符串拆分紅單個的單詞

    */

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { //若是是Scala，因爲SAM轉換，因此能夠寫成val words = lines.flatMap { line => line.split(" ")}

@Override

public Iterable<String> call(String line) throws Exception {

return Arrays.asList(line.split(" "));

}

});

/*

      * 第四步：對初始的DStream進行Transformation級別的處理，例如map、filter等高階函數等的編程，來進行具體的數據計算

      * 第4.2步：在單詞拆分的基礎上對每一個單詞實例計數爲1，也就是word => (word, 1)

      */

JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

@Override

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

/*

      * 第四步：對初始的DStream進行Transformation級別的處理，例如map、filter等高階函數等的編程，來進行具體的數據計算

      *第4.3步：在這裏是經過updateStateByKey來以Batch Interval爲單位來對歷史狀態進行更新，

      * 這是功能上的一個很是大的改進，不然的話須要完成一樣的目的，就可能須要把數據保存在Redis、

      * Tagyon或者HDFS或者HBase或者數據庫中來不斷的完成一樣一個key的State更新，若是你對性能有極爲苛刻的要求，

      * 且數據量特別大的話，能夠考慮把數據放在分佈式的Redis或者Tachyon內存文件系統中；

      * 固然從Spark1.6.x開始能夠嘗試使用mapWithState，Spark2.X後mapWithState應該很是穩定了。

 */

JavaPairDStream<String, Integer> wordsCount = pairs.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { //對相同的Key，進行Value的累計（包括Local和Reducer級別同時Reduce）

@Override

public Optional<Integer> call(List<Integer> values, Optional<Integer> state)

throws Exception {

Integer updatedValue = 0 ;

if(state.isPresent()){

updatedValue = state.get();

}

for(Integer value: values){

updatedValue += value;

}

return Optional.of(updatedValue);

}

});

/*

*此處的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

*主要就是用用來完成這些功能的，並且能夠隨意的自定義具體數據到底放在哪裏！！！

*/

wordsCount.print();

/*

* Spark Streaming執行引擎也就是Driver開始運行，Driver啓動的時候是位於一條新的線程中的，固然其內部有消息循環體，用於

* 接受應用程序自己或者Executor中的消息；

*/

jsc.start();

jsc.awaitTermination();

jsc.close();

}

2.建立checkpoint目錄：

jsc.checkpoint("/usr/local/tmp/checkpoint");

3.　在eclipse中經過run 方法啓動main函數：

4.啓動hdfs服務併發送nc -lk 9999請求：

5.查看checkpoint目錄輸出：

 

源碼解析：

1.PairDStreamFunctions類：

/**

 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of each key.
 * Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
 * @param updateFunc State update function. If `this` function returns None, then
 *                   corresponding state key-value pair will be eliminated.
 * @tparam S State type
 */
def updateStateByKey[S: ClassTag](
    updateFunc: (Seq[V], Option[S]) => Option[S]
  ): DStream[(K, S)] = ssc.withScope {
  updateStateByKey(updateFunc, defaultPartitioner())
}

/**
 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of the key.
 * org.apache.spark.Partitioner is used to control the partitioning of each RDD.
 * @param updateFunc State update function. If `this` function returns None, then
 *                   corresponding state key-value pair will be eliminated.
 * @param partitioner Partitioner for controlling the partitioning of each RDD in the new
 *                    DStream.
 * @tparam S State type
 */
def updateStateByKey[S: ClassTag](
    updateFunc: (Seq[V], Option[S]) => Option[S],
    partitioner: Partitioner
  ): DStream[(K, S)] = ssc.withScope {
  val cleanedUpdateF = sparkContext.clean(updateFunc)
  val newUpdateFunc = (iterator: Iterator[(K, Seq[V], Option[S])]) => {
    iterator.flatMap(t => cleanedUpdateF(t._2, t._3).map(s => (t._1, s)))
  }
  updateStateByKey(newUpdateFunc, partitioner, true)
}

/**
 * Return a new "state" DStream where the state for each key is updated by applying
 * the given function on the previous state of the key and the new values of each key.
 * org.apache.spark.Partitioner is used to control the partitioning of each RDD.
 * @param updateFunc State update function. Note, that this function may generate a different
 *                   tuple with a different key than the input key. Therefore keys may be removed
 *                   or added in this way. It is up to the developer to decide whether to
 *                   remember the partitioner despite the key being changed.
 * @param partitioner Partitioner for controlling the partitioning of each RDD in the new
 *                    DStream
 * @param rememberPartitioner Whether to remember the paritioner object in the generated RDDs.
 * @tparam S State type
 */

def updateStateByKey[S: ClassTag](
    updateFunc: (Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)],
    partitioner: Partitioner,
    rememberPartitioner: Boolean
  ): DStream[(K, S)] = ssc.withScope {
   new StateDStream(self, ssc.sc.clean(updateFunc), partitioner, rememberPartitioner, None)
}

override def compute(validTime: Time): Option[RDD[(K, S)]] = {

  // Try to get the previous state RDD
  getOrCompute(validTime - slideDuration) match {

    case Some(prevStateRDD) => {    // If previous state RDD exists

      // Try to get the parent RDD
      parent.getOrCompute(validTime) match {
        case Some(parentRDD) => {   // If parent RDD exists, then compute as usual
          computeUsingPreviousRDD (parentRDD, prevStateRDD)
        }
        case None => {    // If parent RDD does not exist

          // Re-apply the update function to the old state RDD
          val updateFuncLocal = updateFunc
          val finalFunc = (iterator: Iterator[(K, S)]) => {
            val i = iterator.map(t => (t._1, Seq[V](), Option(t._2)))
            updateFuncLocal(i)
          }
          val stateRDD = prevStateRDD.mapPartitions(finalFunc, preservePartitioning)
          Some(stateRDD)
        }
      }
    }

    case None => {    // If previous session RDD does not exist (first input data)

      // Try to get the parent RDD
      parent.getOrCompute(validTime) match {
        case Some(parentRDD) => {   // If parent RDD exists, then compute as usual
          initialRDD match {
            case None => {
              // Define the function for the mapPartition operation on grouped RDD;
              // first map the grouped tuple to tuples of required type,
              // and then apply the update function
              val updateFuncLocal = updateFunc
              val finalFunc = (iterator : Iterator[(K, Iterable[V])]) => {
                updateFuncLocal (iterator.map (tuple => (tuple._1, tuple._2.toSeq, None)))
              }

              val groupedRDD = parentRDD.groupByKey (partitioner)
              val sessionRDD = groupedRDD.mapPartitions (finalFunc, preservePartitioning)
              // logDebug("Generating state RDD for time " + validTime + " (first)")
              Some (sessionRDD)
            }
            case Some (initialStateRDD) => {
              computeUsingPreviousRDD(parentRDD, initialStateRDD)
            }
          }
        }
        case None => { // If parent RDD does not exist, then nothing to do!
          // logDebug("Not generating state RDD (no previous state, no parent)")
          None
        }
      }
    }
  }

總結：

姜偉

備註：93課

更多私密內容，請關注微信公衆號：DT_Spark