updateStateByKey與mapwithstate怎么實現

這篇文章主要講解了“updateStateByKey與mapwithstate怎么實現”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“updateStateByKey與mapwithstate怎么實現”吧！

updateStateByKey與mapwithstate 這兩個方法在Dstream中是找不到的，他們是通過隱式轉換來進行實現的

由此可以看到，最終是通過PairDStreamFunctions來實現這兩個方法的。

updateStateByKey

newUpdateFunc 方法是在原有基礎上如何進行更新的方法

defaultPartitioner()獲得默認的分區數

如下代碼出現了一個非常關鍵的地方

new StateDStream(self, ssc.sc.clean(updateFunc), partitioner, rememberPartitioner, None)

StateDStream 繼承自Dstream。

stateDStream自會持久化到內存中

里面有一個很總要的方法：如果存在parent RDD 就將執行computeUsingPreviousRDD方法

在該方法中，有一處性能瓶頸的代碼

每次進行更新的時候都會將原有的parentRDD進行cogroup，這樣程序不斷的運行這樣會導致越來越慢！盡量少用改方法！

Mapwithstate

mapWithState方法的返回值是MapWithStateDStream，我們來看看它的實現類

MapWithStateDStreamImpl

最終返回InternalMapWithStateDStream

跟updateStateByKey一樣是持久化在了內存中

persist(StorageLevel.MEMORY_ONLY)

接下來看看每個繼承自Dstream的最重要的方法 compute：

最終操作的是RDD：MapWithStateRDD

RDD中的partition被MapWithStateRDDRecord代表

MapWithStateRDDRecord有伴生對象：中的方法，該方法是對state進行更新操作，不像 updateStateByKey每次都會進cogroup的操作，而是在原有的基礎上進行更新，效率得到了提高！

def updateRecordWithData[K: ClassTag, V: ClassTag, S: ClassTag, E: ClassTag](
    prevRecord: Option[MapWithStateRDDRecord[K, S, E]],
    dataIterator: Iterator[(K, V)],
    mappingFunction: (Time, K, Option[V], State[S]) => Option[E],
    batchTime: Time,
    timeoutThresholdTime: Option[Long],
    removeTimedoutData: Boolean
  ): MapWithStateRDDRecord[K, S, E] = {
    // Create a new state map by cloning the previous one (if it exists) or by creating an empty one
    val newStateMap = prevRecord.map { _.stateMap.copy() }. getOrElse { new EmptyStateMap[K, S]() }

    val mappedData = new ArrayBuffer[E]
    val wrappedState = new StateImpl[S]()

    // Call the mapping function on each record in the data iterator, and accordingly
    // update the states touched, and collect the data returned by the mapping function
    dataIterator.foreach { case (key, value) =>
      wrappedState.wrap(newStateMap.get(key))
      val returned = mappingFunction(batchTime, key, Some(value), wrappedState)
      if (wrappedState.isRemoved) {
        newStateMap.remove(key)
      } else if (wrappedState.isUpdated
          || (wrappedState.exists && timeoutThresholdTime.isDefined)) {
        newStateMap.put(key, wrappedState.get(), batchTime.milliseconds)
      }
      mappedData ++= returned
    }

    // Get the timed out state records, call the mapping function on each and collect the
    // data returned
    if (removeTimedoutData && timeoutThresholdTime.isDefined) {
      newStateMap.getByTime(timeoutThresholdTime.get).foreach { case (key, state, _) =>
        wrappedState.wrapTimingOutState(state)
        val returned = mappingFunction(batchTime, key, None, wrappedState)
        mappedData ++= returned
        newStateMap.remove(key)
      }
    }

    MapWithStateRDDRecord(newStateMap, mappedData)
  }
}

感謝各位的閱讀，以上就是“updateStateByKey與mapwithstate怎么實現”的內容了，經過本文的學習后，相信大家對updateStateByKey與mapwithstate怎么實現這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！