spark streaming测试之三有状态的接收数据
测试思路：首先，使用上篇文章的程序一发送网络数据；
其次，运行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()
}
}
页:
[1]