春城无处不飞花，小白带你侃SparkStreaming(实战应用篇)Alice菌的博客-

写在前面： 博主是一名软件工程系大数据应用开发专业大二的学生，昵称来源于《爱丽丝梦游仙境》中的Alice和自己的昵称。作为一名互联网小白，写博客一方面是为了记录自己的学习历程，一方面是希望能够帮助到很多和自己一样处于起步阶段的萌新。由于水平有限，博客中难免会有一些错误，有纰漏之处恳请各位大佬不吝赐教！个人小站:https://alices.ibilibili.xyz/ , 博客主页:https://alice.blog.csdn.net/
尽管当前水平可能不及各位大佬，但我还是希望自己能够做得更好，因为一天的生活就是一生的缩影。我希望在最美的年华，做最好的自己！

        自上一篇《春城无处不飞花，小白带你侃SparkStreaming(原理引入篇)》结束之后，博主就一直在酝酿着下一篇怎么开始，这不，忙了几天终于也有了下文。

        码字不易，先赞后看，养成习惯!

第三章 Spark Streaming实战

3.1 WordCount

3.1.1. 需求&准备

• 图解
  

• 首先我们在linux服务器上安装nc工具
  nc是netcat的简称，原本是用来设置路由器,我们可以利用它向某个端口发送数据
  yum install -y nc

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

• 发送数据

3.1.2 代码演示

object Streaming01 {   def main(args: Array[String]): Unit = { // 1. 创建SC     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wc")     val sc: SparkContext = new SparkContext(conf) // 设置日志级别     sc.setLogLevel("WARN") // 2. 创建sc 指定【每个批次的时间】     val ssc: StreamingContext = new StreamingContext(sc,Seconds(5)) // 3. 接收数据，并处理     val socketDatas: ReceiverInputDStream[String] = ssc.socketTextStream("node01",9999)      val WCS: DStream[(String, Int)] = socketDatas.flatMap(a=>a.split(" ")).map(a=>(a,1)).reduceByKey(_+_) // 遍历每个RDD WCS.foreachRDD(RDD=>RDD.foreach(println)) // 4. 开始streaming     ssc.start() // 5. 等待关闭     ssc.awaitTermination() } } 

程序运行，我们在命令行窗口下输入一串以空格间隔的字符。例如:，hadoop spark hadoop hive，接着在IDEA的控制台下就能看到类似的信息：

说明SparjStreaming已经接受到9999端口下传递的信息，并做了一个WordCount，将结果显示在了控制台。
        

3.2 updateStateByKey

3.2.1 问题

在上面的那个案例中存在这样一个问题：
每个批次的单词次数都被正确的统计出来，但是结果不能累加！
如果需要累加需要使用updateStateByKey(func)来更新状态。

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

演示效果：
在9999端口分批次发送数据

可以发现每次的结果可以在原有的基础上进行累计统计

3.3 reduceByKeyAndWindow

3.3.1 图解

滑动窗口转换操作的计算过程如下图所示，
我们可以事先设定一个滑动窗口的长度(也就是窗口的持续时间)，并且设定滑动窗口的时间间隔(每隔多长时间执行一次计算)，

比如设置滑动窗口的长度(也就是窗口的持续时间)为24H,设置滑动窗口的时间间隔(每隔多长时间执行一次计算)为1H
那么意思就是：每隔1H计算最近24H的数据

3.2.2 代码演示

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的数据      wordAndCount.print()     ssc.start()//开启     ssc.awaitTermination()//等待优雅停止 } } 

打开端口nc -lk 9999,运行程序，此时并不做任何的数据输入，在等待了几秒之后开始输入字符串。此时可以观察到IDEA的控制台上已经开始对输入数据做了WordCount。

接下来的几秒，增大输入数据的频率，
可以观察到计算的数据量明显在增大，但当我停止输入数据的时候，数据量直接骤减，直到恢复成了程序最开始的模样。

这是为什么呢？
该案例中我设置的窗口长度windowDuration=10，窗口的滑动距离slideDuration=5
那么执行结果就是每隔5s计算最近10s的数据

所以无论你再怎么输入，程序每次也只计算最近10s(如果设置的批次是5秒，也就是两个批次的范围的数据)，所以就会呈现出上述的结果。

需要注意的是：windowDuration和slideDuration必须是batchDuration的倍数

windowDuration=slideDuration:数据不会丢失也不会重复，但可以通过增大Spark Streaming批次的周期替代
windowDuration>slideDuration:数据会重复计算，开发中会使用
windowDuration<slideDuration:数据会丢失，开发中一般不用

        好了，本篇主要讲解的都是基于SparkStreaming的实战基础应用，受益的朋友或对大数据技术感兴趣的伙伴记得关注支持一波(＾Ｕ＾)ノ~ＹＯ