测试思路:

首先,使用上篇文章的程序一发送网络数据;
其次,运行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()
}
}另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。