大数据之sparkStreaming流式处理，接受socket数据，实现单词统计并且每个批次数据结果累加

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
* sparkStreaming流式处理，接受socket数据，实现单词统计并且每个批次数据结果累加
*/
object SparkStreamingTCPTotal {

//newValues 表示当前批次汇总成的(word,1)中相同单词的所有的1
//runningCount 历史的所有相同key的value总和
//newValues:新过来的值
//runningCount:之前保存的状态值
def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
val newCount =runningCount.getOrElse(0)+newValues.sum
Some(newCount)
}

def main(args: Array[String]): Unit = {

//配置sparkConf参数
val sparkConf = new SparkConf().setAppName("SparkStreamingTCPTotal").setMaster("local[2]")
//构建sparkContext对象
val sc = new SparkContext(sparkConf)

sc.setLogLevel("WARN")
//构建StreamingContext对象，每个批处理的时间间隔
val scc = new StreamingContext(sc, Seconds(5))

// scc.checkpoint("./")
//注册一个监听的IP地址和端口 用来收集数据
val lines = scc.socketTextStream("wanghy", 9999)
//切分每一行记录
val words = lines.flatMap(_.split(" "))
//每个单词记为1
val wordAndOne = words.map((_, 1))
//累计统计单词出现的次数
val result = wordAndOne.updateStateByKey(updateFunction)
result.print()
scc.start()
scc.awaitTermination()
}
}

大数据之sparkStreaming流式处理，接受socket数据，实现单词统计并且每个批次数据结果累加

原文：https://www.cnblogs.com/whyuan/p/13080598.html