0
点赞
收藏
分享

微信扫一扫

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)


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

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

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_Spark


文章目录

  • ​​第三章 Spark Streaming实战​​
  • ​​3.1 WordCount​​
  • ​​3.1.1. 需求&准备​​
  • ​​3.1.2 代码演示​​
  • ​​3.2 updateStateByKey​​
  • ​​3.2.1 问题​​
  • ​​3.3 reduceByKeyAndWindow​​
  • ​​3.3.1 图解​​
  • ​​3.2.2 代码演示​​

第三章 Spark Streaming实战

3.1 WordCount

3.1.1. 需求&准备

  • 图解
    春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_SparkSQL_02

  • 首先我们在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的控制台下就能看到类似的信息:

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_云计算/大数据_03

说明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端口分批次发送数据

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_SparkStreaming_04

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

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_SparkSQL_05

3.3 reduceByKeyAndWindow

3.3.1 图解

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

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

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

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

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_Spark_06

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_SparkSQL_07

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。

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_SparkStreaming_08

接下来的几秒,增大输入数据的频率,

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

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_云计算/大数据_09

这是为什么呢?

该案例中我设置的窗口长度windowDuration=10,窗口的滑动距离slideDuration=5

那么执行结果就是每隔5s计算最近10s的数据

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

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

windowDuration=slideDuration​:数据不会丢失也不会重复,但可以通过增大Spark Streaming批次的周期替代

windowDuration>slideDuration​:数据会重复计算,开发中会使用

windowDuration<slideDuration:​数据会丢失,开发中一般不用

        好了,本篇主要讲解的都是基于SparkStreaming的实战基础应用,​受益的朋友或对大数据技术感兴趣的伙伴记得点赞关注支持一波(^U^)ノ~YO

春城无处不飞花,小白带你侃SparkStreaming(实战应用篇)_数据_10



举报

相关推荐

0 条评论