欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

利用mapWithState实现按照首字母统计的有状态的wordCount

程序员文章站 2022-05-11 10:30:48
利用mapWithState算子实现有状态的wordCount,且按照word的第一个字母为key,但是要求输出的格式为(word,1)这样形式的结果 ......

最近在做sparkstreaming整合kafka的时候遇到了一个问题:

可以抽象成这样一个问题:有状态的wordcount,且按照word的第一个字母为key,但是要求输出的格式为(word,1)这样的形式

举例来说:

例如第一批数据为: hello how when hello

则要求输出为:(hello,1) (how,2) (when,1) (hello,3)

第二批数据为: hello how when what hi

则要求输出为: (hello,4) (how,5) (when,2) (what,3) (hi,6)

首先了解一下mapwithstate的常规用法:

ref:

稍微总结一下mapwithstate的几个tips:

  1. mapwithstate是1.6版本之后推出的
  2. 必须设置checkpoint来储存历史数据
  3. mapwithstate和updatestatebykey的区别 : 他们类似,都是有状态dstream操作, 区别在于,updatestatebykey是输出增量数据,随着时间的增加, 输出的数据越来越多,这样会影响计算的效率, 对cpu和内存压力较大.而mapwithstate则输出本批次数据,但是也含有状态更新.
  4. checkpoint的数据会分散存储在不同的分区中, 在进行状态更新时, 首先会对当前 key 做 hash , 再到对应的分区中去更新状态 , 这种方式大大提高了效率.

解决问题的思路:

state中保存状态为(string,int) 元组类型, 其中string为word的全量, 而int为word的计数.

import org.apache.spark.sparkconf
import org.apache.spark.streaming.dstream.mapwithstatedstream
import org.apache.spark.streaming.{seconds, state, statespec, streamingcontext}

object mapwithstateapp {
  def main(args: array[string]): unit = {
    val conf = new sparkconf().setmaster("local[*]").setappname("mapwithstateapp")
    val ssc = new streamingcontext(conf,seconds(5))
    ssc.checkpoint("c:\\users\\hylz\\desktop\\checkpoint")
    val lines = ssc.sockettextstream("192.168.100.11",8888)
    val words = lines.flatmap(_.split(" "))

    def mappingfunc(key: string, value: option[(string, int)], state: state[(string, int)]): (string, int) = {
      val cnt: int = value.getorelse((null, 0))._2 + state.getoption.getorelse((null, 0))._2
      val allfield: string = value.getorelse((null, 0))._1
      state.update((allfield, cnt))
      (allfield, cnt)
    }

    val cnt: mapwithstatedstream[string, (string, int), (string, int), (string, int)] = words.map(x => (x.substring(0, 1), (x, 1))).mapwithstate(statespec.function(mappingfunc _))

    cnt.print()
    ssc.start()
    ssc.awaittermination()
  }
}

测试结果如下
利用mapWithState实现按照首字母统计的有状态的wordCount

input: hello how when hello
利用mapWithState实现按照首字母统计的有状态的wordCount

input: hello how when what hi
利用mapWithState实现按照首字母统计的有状态的wordCount