有哪些信誉好的足球投注网站- 1、本文档共9页，可阅读全部内容。
- 2、有哪些信誉好的足球投注网站（book118）网站文档一经付费（服务费），不意味着购买了该文档的版权，仅供个人/单位学习、研究之用，不得用于商业用途，未经授权，严禁复制、发行、汇编、翻译或者网络传播等，侵权必究。
- 3、本站所有内容均由合作方或网友上传，本站不对文档的完整性、权威性及其观点立场正确性做任何保证或承诺！文档内容仅供研究参考，付费前请自行鉴别。如您付费，意味着您自己接受本站规则且自行承担风险，本站不退款、不进行额外附加服务；查看《如何避免下载的几个坑》。如果您已付费下载过本站文档，您可以点击 这里二次下载。
- 4、如文档侵犯商业秘密、侵犯著作权、侵犯人身权等，请点击“版权申诉”（推荐），也可以打举报电话：400-050-0827(电话支持时间：9:00-18:30)。
查看更多
State管理之updateStateByKey和mapWithState解密选编
State管理之updateStateByKey和mapWithState解密
背景：
整个Spark Streaming是按照Batch Duractions划分Job的。但是很多时候我们需要算过去的一天甚至一周的数据，这个时候不可避免的要进行状态管理，而Spark Streaming每个Batch Duractions都会产生一个Job，Job里面都是RDD，所以此时面临的问题就是怎么对状态进行维护？这个时候就需要借助updateStateByKey和mapWithState方法完成核心的步骤。
源码分析：
1. 无论是updateStateByKey还是mapWithState方法在DStream中均没有，但是是通过隐身转换函数实现其功能。
object DStream {
// `toPairDStreamFunctions` was in SparkContext before 1.3 and users had to
// `import StreamingContext._` to enable it. Now we move it here to make the compiler find
// it automatically. However, we still keep the old function in StreamingContext for backward
// compatibility and forward to the following function directly.
implicit def toPairDStreamFunctions[K, V](stream: DStream[(K, V)])
(implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null):
PairDStreamFunctions[K, V] = {
new PairDStreamFunctions[K, V](stream)
}
updateStateByKey：
1. 在PairDStreamFunctions中updateStateByKey具体实现如下：
在已有的历史基础上，updateFunc对历史数据进行更新。该函数的返回值是DStream类型的。
/**
* 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 Sparks default number of partitions.
* @param updateFunc State update function. If `this` function returns None, then
* corresponding state key-value pair will be eliminated.
* @tparam S State type
*/
def updateStateByKey[S: ClassTag](
updateFunc: (Seq[V], Option[S]) = Option[S]
): DStream[(K, S)] = ssc.withScope {
// defaultPartitioner
updateStateByKey(updateFunc, defaultPartitioner())
}
2. defaultPartitioner:
private[streaming] def defaultPartitioner(numPartitions: Int = self.ssc.sc.defaultParallelism) = {
new HashPartitioner(numPartitions)
3. partitioner就是控制RDD的每个patition
/**
* Return a new state DStream where the state for each key is updated by applying
* the given function on the previous stat
文档评论（0）