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程序