spark streaming测试之三有状态的接收数据

测试思路：

首先，使用上篇文章的程序一发送网络数据；

其次，运行spark程序，观察效果。

说明：

1. 这里使用到了更新函数；

2. 使用检查点来保证状态。

sparkStreamingimport 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()  }}