6.SparkStreaming之WordCount(UpdateStateByKey)

Posted wddqy

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了6.SparkStreaming之WordCount(UpdateStateByKey)相关的知识,希望对你有一定的参考价值。

代码:

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object UpdateStateByKeyWordCount {
def main(args: Array[String]): Unit = {
Logger.getLogger("org").setLevel(Level.ERROR)
val conf = new SparkConf().setMaster("local[2]").setAppName("UpdateStateByKeyWordCount")
val ssc = new StreamingContext(conf, Seconds(5))
ssc.checkpoint("checkpointdirectory")
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap( _.split(" ")).map(word => (word, 1))
val result = words.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
var newValue = state.getOrElse(0)
for(value <- values) {
newValue += value
}
Option(newValue)
})
result.print()

ssc.start()
ssc.awaitTermination()
}
}

结果:

技术图片

 

 技术图片

 

 技术图片

以上是关于6.SparkStreaming之WordCount(UpdateStateByKey)的主要内容,如果未能解决你的问题,请参考以下文章

Linux巩固记录 hadoop 2.7.4下自己编译代码并运行MapReduce程序

信息领域热词分析系统--python统计

C语言之基本算法35—数组上三角之积 主对角之积 副对角之积

入门设计模式之汇总篇

计算机领域之父

JavaWeb之Ajax&JSON