在线精品99_中国九九盗摄偷拍偷看_91免费版在线观看_91.app_91高清视频在线_99热最新网站

updateStateByKey与mapwithstate怎么实现

135次阅读
没有评论

共计 2598 个字符,预计需要花费 7 分钟才能阅读完成。

这篇文章主要讲解了“updateStateByKey 与 mapwithstate 怎么实现”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着丸趣 TV 小编的思路慢慢深入,一起来研究和学习“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:

updateStateByKey 与 mapwithstate 怎么实现

最终操作的是 RDD:MapWithStateRDD

RDD 中的 partition 被 MapWithStateRDDRecord 代表

updateStateByKey 与 mapwithstate 怎么实现

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

正文完
 
丸趣
版权声明:本站原创文章,由 丸趣 2023-08-16发表,共计2598字。
转载说明:除特殊说明外本站除技术相关以外文章皆由网络搜集发布,转载请注明出处。
评论(没有评论)
主站蜘蛛池模板: 韩国主播19福利视频在线观看 | 2021国产在线视频 | 一级毛片全部免费播放 | 四虎一影院区永久精品 | 男女色网 | 99在线视频网站 | 麻豆精品永久免费视频 | 成人夜色香网站在线观看 | 亚洲另类天堂 | 久久久久国产精品免费免费不卡 | 特黄大片aaaaa毛片 | 免费人成网站在线观看欧美 | 综合欧美一区二区三区 | 久久精品亚洲欧美va | 特级a做爰全过程片 | 欧美成人极品怡红院tv | 26uuu色噜噜欧美在线播放 | 偷自拍视频区综合视频区 | 色猫咪av在线网址 | 少妇人妻无码专区视频 | 国内精品久久久久影院日本 | 国产精品免费视频播放 | 2022国内精品免费福利视频 | 高清毛片在线看高清 | 亚洲精品乱码久久久久久金桔影视 | 亚洲国产视频网 | 亚洲av无码一区二区三区四区 | 亚洲精品一区二区三区四区五区 | 国产高清三级 | 久久久久久午夜精品 | 国产丝袜在线精品丝袜 | 狼人无码精华av午夜精品 | a级做爰视频免费观看 | 久久成人国产精品 | 男女啪啪无遮挡免费网站 | 四虎免费久久影院 | 国产成人亚洲日韩欧美 | 国产永久免费高清在线观看 | 久久er热在这里只有精品85 | 97国产在线播放 | 一级做受毛片免费大片 |