Spark –Spark Streaming实战 WordCount他来啦！！！ updateStateByKey reduceByKeyAndWindow网络kismet-

WordCount

 需求&准备

●图解

●首先在linux服务器上安装nc工具

nc是netcat的简称，原本是用来设置路由器,我们可以利用它向某个端口发送数据

yum install -y nc

●启动一个服务端并开放9999端口,等一下往这个端口发数据

nc -lk 9999

●发送数据

话不多说！！！上代码！

 package cn.itcast.streaming  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.streaming.{Seconds, StreamingContext}  object WordCount {   def main(args: Array[String]): Unit = {     //1.创建StreamingContext     //spark.master should be set as local[n], n > 1     val conf = new SparkConf().setAppName("wc").setMaster("local[*]")     val sc = new SparkContext(conf)     sc.setLogLevel("WARN")     val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD     //2.监听Socket接收数据     //ReceiverInputDStream就是接收到的所有的数据组成的RDD,封装成了DStream,接下来对DStream进行操作就是对RDD进行操作     val dataDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)     //3.操作数据     val wordDStream: DStream[String] = dataDStream.flatMap(_.split(" "))     val wordAndOneDStream: DStream[(String, Int)] = wordDStream.map((_,1))     val wordAndCount: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(_+_)     wordAndCount.print()     ssc.start()//开启     ssc.awaitTermination()//等待停止   } }

 执行

1.先执行nc -lk 9999

2.然后执行代码

3.不断的在1中输入不同的单词

hadoop spark sqoop hadoop spark hive hadoop

4.观察IDEA控制台输出

sparkStreaming每隔5s计算一次当前5s内的数据，然后将每个批次的数据输出

updateStateByKey

​​​​​​​问题

在上面的那个案例中存在这样一个问题：

每个批次的单词次数都被正确的统计出来，但是结果不能累加！

如果需要累加需要使用updateStateByKey(func)来更新状态.

话不多说！！！再上代码！

 package cn.itcast.streaming  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.{SparkConf, SparkContext}   object WordCount2 {   def main(args: Array[String]): Unit = {     //1.创建StreamingContext     //spark.master should be set as local[n], n > 1     val conf = new SparkConf().setAppName("wc").setMaster("local[*]")     val sc = new SparkContext(conf)     sc.setLogLevel("WARN")     val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD     //requirement failed: ....Please set it by StreamingContext.checkpoint().     //注意:我们在下面使用到了updateStateByKey对当前数据和历史数据进行累加     //那么历史数据存在哪?我们需要给他设置一个checkpoint目录     ssc.checkpoint("./wc")//开发中HDFS     //2.监听Socket接收数据     //ReceiverInputDStream就是接收到的所有的数据组成的RDD,封装成了DStream,接下来对DStream进行操作就是对RDD进行操作     val dataDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)     //3.操作数据     val wordDStream: DStream[String] = dataDStream.flatMap(_.split(" "))     val wordAndOneDStream: DStream[(String, Int)] = wordDStream.map((_,1))     //val wordAndCount: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(_+_)     //====================使用updateStateByKey对当前数据和历史数据进行累加====================     val wordAndCount: DStream[(String, Int)] =wordAndOneDStream.updateStateByKey(updateFunc)     wordAndCount.print()     ssc.start()//开启     ssc.awaitTermination()//等待优雅停止   }   //currentValues:当前批次的value值,如:1,1,1 (以测试数据中的hadoop为例)   //historyValue:之前累计的历史值,第一次没有值是0,第二次是3   //目标是把当前数据+历史数据返回作为新的结果(下次的历史数据)   def updateFunc(currentValues:Seq[Int], historyValue:Option[Int] ):Option[Int] ={ // currentValues当前值 // historyValue历史值     val result: Int = currentValues.sum + historyValue.getOrElse(0)     Some(result)   } }

​​​​​​​执行

1.先执行nc -lk 9999

2.然后执行以上代码

3.不断的在1中输入不同的单词，

hadoop spark sqoop hadoop spark hive hadoop

4.观察IDEA控制台输出

sparkStreaming每隔5s计算一次当前5s内的数据，然后将每个批次的结果数据累加输出。

 reduceByKeyAndWindow

​​​​​​​ 图解

滑动窗口转换操作的计算过程如下图所示，

我们可以事先设定一个滑动窗口的长度(也就是窗口的持续时间)，并且设定滑动窗口的时间间隔(每隔多长时间执行一次计算)，

比如设置滑动窗口的长度(也就是窗口的持续时间)为24H,设置滑动窗口的时间间隔(每隔多长时间执行一次计算)为1H

那么意思就是:每隔1H计算最近24H的数据

代码解决一切！！！再上代码！

 package cn.itcast.streaming  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.{SparkConf, SparkContext}  object WordCount3 {   def main(args: Array[String]): Unit = {     //1.创建StreamingContext     //spark.master should be set as local[n], n > 1     val conf = new SparkConf().setAppName("wc").setMaster("local[*]")     val sc = new SparkContext(conf)     sc.setLogLevel("WARN")     val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD     //2.监听Socket接收数据     //ReceiverInputDStream就是接收到的所有的数据组成的RDD,封装成了DStream,接下来对DStream进行操作就是对RDD进行操作     val dataDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)     //3.操作数据     val wordDStream: DStream[String] = dataDStream.flatMap(_.split(" "))     val wordAndOneDStream: DStream[(String, Int)] = wordDStream.map((_,1))       val wordAndCount: DStream[(String, Int)] = wordAndOneDStream.reduceByKeyAndWindow((a:Int,b:Int)=>a+b,Seconds(10),Seconds(5))    //4.使用窗口函数进行WordCount计数     //reduceFunc: (V, V) => V,集合函数     //windowDuration: Duration,窗口长度/宽度     //slideDuration: Duration,窗口滑动间隔     //注意:windowDuration和slideDuration必须是batchDuration的倍数     //windowDuration=slideDuration:数据不会丢失也不会重复计算==开发中会使用     //windowDuration>slideDuration:数据会重复计算==开发中会使用     //windowDuration<slideDuration:数据会丢失     //代码表示:     //windowDuration=10     //slideDuration=5     //那么执行结果就是每隔5s计算最近10s的数据     //比如开发中让你统计最近1小时的数据,每隔1分钟计算一次,那么参数该如何设置?      wordAndCount.print()     ssc.start()//开启     ssc.awaitTermination()//等待优雅停止   } }

​​​​​​​执行

1.先执行nc -lk 9999

2.然后执行以上代码

3.不断的在1中输入不同的单词

hadoop spark sqoop hadoop spark hive hadoop

4.观察IDEA控制台输出

现象：sparkStreaming每隔5s计算一次当前在窗口大小为10s内的数据，然后将结果数据输出。