SparkStreaming

Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的 API。支持kafka，flume等大量数据源。

准实时，微批次处理。

可以理解为一个while（true）的循环启动在Executor上，一直二十四小时执行。随时等着数据过来，然后把数据分解发送给Driver，然后使用spark对数据进行处理。但是底层并不是过来多少数据处理多少数据，毕竟内存是有限的。所以一般通过设置将一段的时间段的数据来处理，每次处理某个采集周期的数据。

DStream (discretized stream离散化流)是SparkStreaming的一个基于RDD抽象出来的一个类似概念。可以直接当做RDD理解。

image.png

概述

特点

1. 易用

2. 容错

3. 易整合到spark体系

简单使用

启动SparkStreaming 监听9999端口数据

//1.初始化 Spark 配置信息
val sparkConf = new
SparkConf().setMaster("local[*]").setAppName("StreamWordCount")
//2.初始化 SparkStreamingContext （spark配置，批量处理周期3s）
val ssc = new StreamingContext(sparkConf, Seconds(3))
//3.通过监控端口创建 DStream，读进来的数据为一行行
val lineStreams = ssc.socketTextStream("linux1", 9999)
//将每一行数据做切分，形成一个个单词
val wordStreams = lineStreams.flatMap(_.split(" "))
//将单词映射成元组(word,1)
val wordAndOneStreams = wordStreams.map((_, 1))
//将相同的单词次数做统计
val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)
//打印 wordAndCountStreams.print()
//启动 SparkStreamingContext ssc.start() ssc.awaitTermination()
}

DStream使用

基本与RDD使用上相同，命名可能有些许变化。

有状态的操作

类似于Spark中的Acc累加器，需要把一段时间的数据累计起来使用。（正常使用的算子一般为无状态，只处理一次）

image.png

 val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
 val ssc = new StreamingContext(sparkConf, Seconds(3))

 //注意：需要设置检查点
 ssc.checkpoint("cp")

 // 无状态数据操作，只对当前的采集周期内的数据进行处理
 // 在某些场合下，需要保留数据统计结果（状态），实现数据的汇总
 // 使用有状态操作时，需要设定检查点路径
 val datas = ssc.socketTextStream("localhost", 9999)

 val wordToOne = datas.map((_,1))

 //val wordToCount = wordToOne.reduceByKey(_+_)

 // updateStateByKey：根据key对数据的状态进行更新
 // 传递的参数中含有两个值
 // 第一个值表示相同的key的value数据
 // 第二个值表示缓存区相同key的value数据
 val state = wordToOne.updateStateByKey(
 //seq：当前数据，//buff：缓冲区中的数据（之前处理数据的结果）
 ( seq:Seq[Int], buff:Option[Int] ) => {
 val newCount = buff.getOrElse(0) + seq.sum
 Option(newCount)
 }
 )

 //会将从任务开始 获取的所有数据的word count统计得到
 state.print()

 ssc.start()
 ssc.awaitTermination()
 }

Transform

DStream支持的api不能够满足使用，可以转换成RDD来调用更多的api

但是要注意执行位置的区别（RDD内才是在Driver执行）

 // transform方法可以将底层RDD获取到后进行操作
 // 1\. DStream功能不完善
 // 2\. 需要代码周期性的执行

 // Code : Driver端
 val newDS: DStream[String] = lines.transform(
 //DStream -> RDD (RDD支持更多便利的api)
 rdd => {
 val rdd1: RDD[String] = rdd 
 // Code : Driver端，（周期性执行）
 rdd.map(
 str => {
 // Code : Executor端
 str
 }
 )
 }
 )
 // Code : Driver端
 val newDS1: DStream[String] = lines.map(
 data => {
 // Code : Executor端
 data
 }
 )

Window

每达到一个streaming的timing，window就会往前移一格，查看是否达到了时间，如果达到了，就会对window框选的数据进行处理。

当然也可以调整window计算的步长，比如让window每60s计算一次，就可以避开计算重复数据。

image.png

基本使用

解决需求1：显示每分钟的数据处理性能。

 val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
 val ssc = new StreamingContext(sparkConf, Seconds(10))

 val lines = ssc.socketTextStream("localhost", 9999)
 val wordToOne = lines.map((_,1))

 // 窗口的范围应该是采集周期的整数倍
 // 窗口可以滑动的，但是默认情况下，一个采集周期进行滑动
 // 每三秒处理一次数据，每10秒计算一次 ,注意是每10秒 ，这6个10秒所处理的数据信息
 val windowDS: DStream[(String, Int)] = wordToOne.window(Seconds(60))

 val wordToCount = windowDS.reduceByKey(_+_)

 wordToCount.print()

 ssc.start()
 ssc.awaitTermination()
 }
}

解决需求2：每分钟统计一次数据总和。

// 每60秒计算一次，最近60秒处理的数据信息 不会有重复计算ßßß
val windowDS: DStream[(String, Int)] = wordToOne.window(Seconds(60),Seconds(60))

image.png

算子

window(windowLength, slideInterval)

基于对源 DStream 窗化的批次进行计算返回一个新的 Dstream;

countByWindow(windowLength, slideInterval):

返回一个滑动窗口计数流中的元素个数;

reduceByWindow(func, windowLength, slideInterval)

通过使用自定义函数整合滑动区间流元素来创建一个新的单元素流;

reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])

当在一个(K,V)对的 DStream 上调用此函数，会返回一个新(K,V)对的 DStream，此处通过对滑动窗口中批次数 据使用 reduce 函数来整合每个 key 的 value 值。

reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])

这个函 数是上述函数的变化版本，每个窗口的 reduce 值都是通过用前一个窗的 reduce 值来递增计算。 通过 reduce 进入到滑动窗口数据并”反向 reduce”离开窗口的旧数据来实现这个操作。一个例 子是随着窗口滑动对 keys 的“加”“减”计数。通过前边介绍可以想到，这个函数只适用于”可逆的 reduce 函数”，也就是这些 reduce 函数有相应的”反 reduce”函数(以参数 invFunc 形式 传入)。如前述函数，reduce 任务的数量通过可选参数来配置。

val ssc = new StreamingContext(sparkConf, Seconds(3))
 //有状态计算 需要开启检查点
 ssc.checkpoint("cp")

 //("a",1)
 val wordToOne = lines.map((_,1))

 //在窗口数据比较多 对其进行新增和减少 要比重新计算的性能要高
 val windowDS: DStream[(String, Int)] =
 wordToOne.reduceByKeyAndWindow(
 //加上窗口滑动增加的数据 x当前数据 y新进窗口的数据
 (x:Int, y:Int) => { x + y},
 //减去窗口滑动减少的数据 x当前数据 y新离开窗口的数据
 (x:Int, y:Int) => {x - y},
 Seconds(9), Seconds(3))
 // SparkStreaming如何没有输出操作，那么会提示错误
 //windowDS.print()

Output

foreachRDD(func) 最通用的输出操作

 //计算结果
 val windowDS: DStream[(String, Int)] =
 wordToOne.reduceByKeyAndWindow(
 (x:Int, y:Int) => { x + y},
 (x:Int, y:Int) => {x - y},
 Seconds(9), Seconds(3))


 // foreachRDD不会出现时间戳
 windowDS.foreachRDD(
 rdd => {

 }
 )

saveAsObjectFiles(prefix, [suffix])

以 Java 对象序列化的方式将 Stream 中的数据保存为 SequenceFiles . 每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]". Python 中目前不可用。

saveAsHadoopFiles(prefix, [suffix])

将 Stream 中的数据保存为 Hadoop files. 每一批次的存 储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]"。Python API 中目前不可用。