這篇文章主要講解了“updateStateByKey與mapwithstate怎么實(shí)現(xiàn)”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“updateStateByKey與mapwithstate怎么實(shí)現(xiàn)”吧！

成都創(chuàng)新互聯(lián)-專業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設(shè)、高性價(jià)比清徐網(wǎng)站開發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫(kù),直接使用。一站式清徐網(wǎng)站制作公司更省心,省錢,快速模板網(wǎng)站建設(shè)找我們，業(yè)務(wù)覆蓋清徐地區(qū)。費(fèi)用合理售后完善，十多年實(shí)體公司更值得信賴。

updateStateByKey與mapwithstate 這兩個(gè)方法在Dstream中是找不到的，他們是通過隱式轉(zhuǎn)換來進(jìn)行實(shí)現(xiàn)的

由此可以看到，最終是通過PairDStreamFunctions來實(shí)現(xiàn)這兩個(gè)方法的。

updateStateByKey

newUpdateFunc 方法是在原有基礎(chǔ)上如何進(jìn)行更新的方法

defaultPartitioner()獲得默認(rèn)的分區(qū)數(shù)

如下代碼出現(xiàn)了一個(gè)非常關(guān)鍵的地方

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

StateDStream 繼承自Dstream。

stateDStream自會(huì)持久化到內(nèi)存中

里面有一個(gè)很總要的方法：如果存在parent RDD 就將執(zhí)行computeUsingPreviousRDD方法

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

每次進(jìn)行更新的時(shí)候都會(huì)將原有的parentRDD進(jìn)行cogroup，這樣程序不斷的運(yùn)行這樣會(huì)導(dǎo)致越來越慢！盡量少用改方法！

Mapwithstate

mapWithState方法的返回值是MapWithStateDStream，我們來看看它的實(shí)現(xiàn)類

MapWithStateDStreamImpl

最終返回InternalMapWithStateDStream

跟updateStateByKey一樣是持久化在了內(nèi)存中

persist(StorageLevel.MEMORY_ONLY)

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

最終操作的是RDD：MapWithStateRDD

RDD中的partition被MapWithStateRDDRecord代表

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

defupdateRecordWithData[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
    valnewStateMap = prevRecord.map { _.stateMap.copy() }. getOrElse { newEmptyStateMap[K, S]() }

    valmappedData = newArrayBuffer[E]
    valwrappedState = newStateImpl[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))
      valreturned = 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)
        valreturned = mappingFunction(batchTime, key, None, wrappedState)
        mappedData ++= returned
        newStateMap.remove(key)
      }
    }

    MapWithStateRDDRecord(newStateMap, mappedData)
  }
}

感謝各位的閱讀，以上就是“updateStateByKey與mapwithstate怎么實(shí)現(xiàn)”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對(duì)updateStateByKey與mapwithstate怎么實(shí)現(xiàn)這一問題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！