测试思路：
首先，使用上篇文章的程序一发送网络数据；
其次，运行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()
}
}免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
