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怎么实现这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！