测试思路:
首先,使用上篇文章的程序一发送网络数据;
其次,运行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() } }
本文出自 “一步.一步” 博客,转载请与作者联系!
原文地址:http://snglw.blog.51cto.com/5832405/1656794