测试思路:
网站建设哪家好,找成都创新互联!专注于网页设计、网站建设、微信开发、微信小程序定制开发、集团企业网站建设等服务项目。为回馈新老客户创新互联还提供了龙潭免费建站欢迎大家使用!
首先,使用上篇文章的程序一发送网络数据;
其次,运行spark程序,观察效果。
说明:
1. 这里使用到了更新函数;
2. 使用检查点来保证状态。
sparkStreaming import org.apache.log4j.{LoggerLevel} import org.apache.spark.streaming.{SecondsStreamingContext} import org.apache.spark.{SparkContextSparkConf} import org.apache.spark.streaming.StreamingContext._ object StatefulWordCount { def main(args:Array[]){ Logger.().setLevel(Level.) Logger.().setLevel(Level.) updateFunc = (values: []state:Option[]) => { currentCount = values.foldLeft()(_+_) previousCount = state.getOrElse() (currentCount + previousCount) } conf = SparkConf().setAppName().setMaster() sc = SparkContext(conf) ssc = StreamingContext(sc()) ssc.checkpoint() lines = ssc.socketTextStream(args()args().toInt) words = lines.flatMap(_.split()) wordCounts = words.map(x=>(x)) stateDstream = wordCounts.updateStateByKey[](updateFunc) stateDstream.print() ssc.start() ssc.awaitTermination() } }