扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
测试思路：
网站建设哪家好，找成都创新互联！专注于网页设计、网站建设、微信开发、微信小程序定制开发、集团企业网站建设等服务项目。为回馈新老客户创新互联还提供了龙潭免费建站欢迎大家使用！
首先，使用上篇文章的程序一发送网络数据；
其次，运行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()
}
}
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
微信二维码
移动版官网