在面向流處理的分布式計(jì)算中，經(jīng)常會(huì)有這種需求，希望需要處理的某個(gè)數(shù)據(jù)集能夠不隨著流式數(shù)據(jù)的流逝而消失。

以spark streaming為例固灵，就是希望有個(gè)數(shù)據(jù)集能夠在當(dāng)前批次中更新，再下個(gè)批次后又可以繼續(xù)訪問(wèn)劫流。一個(gè)最簡(jiǎn)單的實(shí)現(xiàn)是在driver的內(nèi)存中巫玻，我們可以自行保存一個(gè)大的內(nèi)存結(jié)構(gòu)。這種hack的方式就是我們無(wú)法利用spark提供的分布式計(jì)算的能力困介。

對(duì)此大审，spark streaming提供了stateful streaming, 可以創(chuàng)建一個(gè)有狀態(tài)的DStream蘸际，我們可以操作一個(gè)跨越不同批次的RDD座哩。

1 updateStateByKey

該方法提供了這樣的一種機(jī)制： 維護(hù)了一個(gè)可以跨越不同批次的RDD， 姑且成為StateRDD粮彤，在每個(gè)批次遍歷StateRDD的所有數(shù)據(jù)根穷，對(duì)每條數(shù)據(jù)執(zhí)行update方法。當(dāng)update方法返回None時(shí)导坟，淘汰StateRDD中的該條數(shù)據(jù)屿良。

具體接口如下：

/**

* 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 `numPartitions` partitions.

* @param updateFunc State update function. If `this` function returns None, then

*                   corresponding state key-value pair will be eliminated.

* @param numPartitions Number of partitions of each RDD in the new DStream.

* @tparam S State type

*/

def updateStateByKey[S: ClassTag](

    updateFunc: (Seq[V], Option[S]) => Option[S],

    numPartitions: Int

  ): DStream[(K, S)] = ssc.withScope {

  updateStateByKey(updateFunc, defaultPartitioner(numPartitions))

}

即用戶需要實(shí)現(xiàn)一個(gè)updateFunc的函數(shù)，該函數(shù)的參數(shù)：

Seq[V] 該批次中相同key的數(shù)據(jù)惫周，以Seq數(shù)組形式傳遞

Option[S] 歷史狀態(tài)中的數(shù)據(jù)

返回值： 返回需要保持的歷史狀態(tài)數(shù)據(jù)尘惧，為None時(shí)表示刪除該數(shù)據(jù)

def updateStateFunc(lines: Seq[Array[String]], state: Option[Array[String]]): Option[Array[String]] = {...}

這種做法簡(jiǎn)單清晰明了，但是其中有一些可以優(yōu)化的地方：

a) 如果DRDD增長(zhǎng)到比較大的時(shí)候递递，而每個(gè)進(jìn)入的批次數(shù)據(jù)量相比并不大喷橙，此時(shí)每次都需要遍歷DRDD，無(wú)論該批次中是否有數(shù)據(jù)需要更新DRDD登舞。這種情況有的時(shí)候可能會(huì)引發(fā)性能問(wèn)題贰逾。

b) 需要用戶自定義數(shù)據(jù)的淘汰機(jī)制。有的時(shí)候顯得不是那么方便菠秒。

c) 返回的類型需要和緩存中的類型相同疙剑。類型不能發(fā)生改變。

2 mapWithState

該接口是對(duì)updateSateByKey的改良，解決了updateStateFunc中可以優(yōu)化的地方：

/**

* :: Experimental ::

* Return a [[MapWithStateDStream]] by applying a function to every key-value element of

* `this` stream, while maintaining some state data for each unique key. The mapping function

* and other specification (e.g. partitioners, timeouts, initial state data, etc.) of this

* transformation can be specified using [[StateSpec]] class. The state data is accessible in

* as a parameter of type [[State]] in the mapping function.

*

* Example of using `mapWithState`:

* {{{

*    // A mapping function that maintains an integer state and return a String

*    def mappingFunction(key: String, value: Option[Int], state: State[Int]): Option[String] = {

*      // Use state.exists(), state.get(), state.update() and state.remove()

*      // to manage state, and return the necessary string

*    }

*

*    val spec = StateSpec.function(mappingFunction).numPartitions(10)

*

*    val mapWithStateDStream = keyValueDStream.mapWithState[StateType, MappedType](spec)

* }}}

*

* @param spec          Specification of this transformation

* @tparam StateType    Class type of the state data

* @tparam MappedType   Class type of the mapped data

*/

@Experimental

def mapWithState[StateType: ClassTag, MappedType: ClassTag](

    spec: StateSpec[K, V, StateType, MappedType]

  ): MapWithStateDStream[K, V, StateType, MappedType] = {

  new MapWithStateDStreamImpl[K, V, StateType, MappedType](

    self,

    spec.asInstanceOf[StateSpecImpl[K, V, StateType, MappedType]]

  )

}

其中spec封裝了用戶自定義的函數(shù)言缤，用以更新緩存數(shù)據(jù)：

mappingFunction: (KeyType, Option[ValueType], State[StateType]) => MappedType

實(shí)現(xiàn)樣例如下：

val mappingFunc = (k: String, line: Option[Array[String]], state: State[Array[String]]) => {...}

參數(shù)分別代表：

數(shù)據(jù)的key: k

RDD中的每行數(shù)據(jù): line

state: 緩存數(shù)據(jù)

當(dāng)對(duì)state調(diào)用remove方法時(shí)嚼蚀，該數(shù)據(jù)會(huì)被刪除。

注意管挟，如果數(shù)據(jù)超時(shí)驰坊，不要調(diào)用remove方法，因?yàn)閟park會(huì)在mappingFunc后自動(dòng)調(diào)用remove哮独。

a) 與updateStateByKey 每次都要遍歷緩存數(shù)據(jù)不同拳芙，mapWithState每次遍歷每個(gè)批次中的數(shù)據(jù)，更新緩存中的數(shù)據(jù)皮璧。對(duì)于緩存數(shù)據(jù)較大的情況來(lái)說(shuō)舟扎，性能會(huì)有較大提升。

b) 提供了內(nèi)置的超時(shí)機(jī)制悴务，當(dāng)數(shù)據(jù)一定時(shí)間內(nèi)沒(méi)有更新時(shí)睹限，淘汰相應(yīng)數(shù)據(jù)。

注意讯檐，當(dāng)有數(shù)據(jù)到來(lái)或者有超時(shí)發(fā)生時(shí)羡疗，mappingFunc都會(huì)被調(diào)用。

3 checkpointing

通常情況下别洪，在一個(gè)DStream鐘叨恨，對(duì)RDD的各種轉(zhuǎn)換而依賴的數(shù)據(jù)都是來(lái)自于當(dāng)前批次中。但是當(dāng)在進(jìn)行有狀態(tài)的transformations時(shí)挖垛，包括updateStateByKey/reduceByKeyAndWindow 痒钝、mapWithSate，還會(huì)依賴于以前批次的數(shù)據(jù)痢毒，RDD的容錯(cuò)機(jī)制送矩，在異常情況需要重新計(jì)算RDD時(shí)，需要以前批次的RDD信息哪替。如果這個(gè)依賴的鏈路過(guò)長(zhǎng)栋荸，會(huì)需要大量的內(nèi)存，即使有些RDD的數(shù)據(jù)在內(nèi)存中凭舶，不需要計(jì)算晌块。此時(shí)spark通過(guò)checkpoint來(lái)打破依賴鏈路。checkpoint會(huì)生成一個(gè)新的RDD到hdfs中库快，該RDD是計(jì)算后的結(jié)果集摸袁，而沒(méi)有對(duì)之前的RDD依賴。

此時(shí)一定要啟用checkpointing义屏，以進(jìn)行周期性的RDD Checkpointing

在StateDstream在實(shí)現(xiàn)RDD的compute方法時(shí)靠汁，就是將之前的PreStateRDD與當(dāng)前批次中依賴的ParentRDD進(jìn)行合并蜂大。

而checkpoint的實(shí)現(xiàn)是將上述合并的RDD寫(xiě)入HDFS中。

現(xiàn)在checkpoint的實(shí)現(xiàn)中蝶怔，數(shù)據(jù)寫(xiě)入hdfs的過(guò)程是由一個(gè)固定的線程池異步完成的奶浦。一種存在的風(fēng)險(xiǎn)是上次checkpoint的數(shù)據(jù)尚未完成，此次又來(lái)了新的要寫(xiě)的checkpoint數(shù)據(jù)踢星，會(huì)加大集群的負(fù)載澳叉，可能會(huì)引發(fā)一系列的問(wèn)題。

4 checkpoint周期設(shè)置：

對(duì)mapWithStateByKey/updateStateByKey返回的DStream可以調(diào)用checkpoint方法設(shè)置checkpoint的周期沐悦。注意傳遞的時(shí)間只能是批次時(shí)間的整數(shù)倍成洗。

另外，對(duì)于mapWithState而言藏否，checkpoint執(zhí)行時(shí)瓶殃，才會(huì)進(jìn)行數(shù)據(jù)的刪除。 State.remove方法只是設(shè)置狀態(tài)副签，標(biāo)記為刪除遥椿，數(shù)據(jù)并不會(huì)真的刪除。 SnapShot方法還是可以獲取得到淆储。

參考:
[1] https://halfvim.github.io/2016/06/19/Checkpointing-in-Spark-Streaming/
[2] http://asyncified.io/2016/07/31/exploring-stateful-streaming-with-apache-spark/