import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}import org.apache.spark.SparkConfval sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]")//10秒创建一个批次val ssc = new StreamingContext(sparkConf, Seconds(10))val lines = ssc.socketTextStream("localhost", 9999)val wordCounts = lines.flatMap(_.split(" ")).map(x => (x, 1l))val current_stream = wordCounts.reduceByKey(_ + _)current_stream.print()current_stream.repartition(1).saveAsTextFiles("/data/socket_wordcount_current.")ssc.start()ssc.awaitTermination()

import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}
import org.apache.spark.SparkConf

val sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]")//10秒创建一个批次val ssc = new StreamingContext(sparkConf, Seconds(10))
//累加所有经过的数据val updateFunc = (values: Seq[Long], state: Option[Long]) => {val currentCount = values.foldLeft(0l)(_ + _)val previousCount = state.getOrElse(0l)Some(currentCount + previousCount)}ssc.checkpoint("socket_wordcount_history")val lines = ssc.socketTextStream("localhost", 9999)val wordCounts = lines.flatMap(_.split(" ")).map(x => (x, 1l))val history_stream = wordCounts.updateStateByKey[Long](updateFunc)//合并当前数据和历史数据history_stream.print()history_stream.repartition(1).saveAsTextFiles("/data/socket_wordcount_history.")ssc.start()ssc.awaitTermination()

sc.stopimport org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}import org.apache.spark.SparkConfval updateFunc = (values: Seq[Long], state: Option[Long]) => {
val currentCount = values.foldLeft(0l)(_ + _)
val previousCount = state.getOrElse(0l)
Some(currentCount + previousCount)
}
val sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(10))   //10秒创建一个批次
ssc.checkpoint("socket-kafka-wordcount_recent")
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1l))
val stateDstream = wordCounts.reduceByKeyAndWindow(_ + _, _ - _, Minutes(1), Seconds(30))  //每30秒算一次，数据范围为最近一分钟内收到的数据  另外，使用window时，需要设置checkpointstateDstream.print()
stateDstream.repartition(1).saveAsTextFiles("/data/socket-streaming-wordcount.log")ssc.start()
ssc.awaitTermination()