SparkStreaming DStream转换
newfarhui 2019-12-19
1、无状态转换操作
（1）无状态转化操作就是把简单的RDD转化操作应用到每个批次上，也就是转换DStream中的每一个RDD。
部分无状态转化操作：
（2）尽管这些函数韩起来像作用在整个流上一样，但事实上每个DStream在内部是由许多RDD（批次）组成，且无状态转化操作是分别应用到每个RDD上。
例如：reduceByKey（）会化简每个事件区间中的数据，但不会化简不同区间之间的数据。
（3）在wordcount中，我们只会统计几秒内接收到的数据的单词个数，而不会累加
（4）无状态转化操作也能在多个DStream间整合数据，不过也是在各个事件区间内。如，键值对DStream拥有和RDD一样的与连接相关的转化操作，也就是cogroup()、join()、leftOuterJoin() 等。我们可以在DStream上使用这些操作，这样就对每个批次分别执行了对应的RDD操作。
（5）我们还可以像在常规的 Spark 中一样使用 DStream的union() 操作将它和另一个DStream 的内容合并起来，也可以使用StreamingContext.union()来合并多个流。
transform操作
transform允许DStream上执行任意的RDD-to-RDD函数
可以用来执行一些 RDD 操作, 即使这些操作并没有在 SparkStreaming 中暴露出来.
该函数每一批次调度一次。其实也就是对DStream中的RDD应用转换。
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Author z
* Date 2019/4/28 6:51 PM
*/
object TransformDemo {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Practice").setMaster("local[2]")
val sctx = new StreamingContext(conf, Seconds(3))
val dstream: ReceiverInputDStream[String] = sctx.socketTextStream("hadoop201", 10000)
val resultDStream = dstream.transform(rdd => {
rdd.flatMap(_.split("\\W")).map((_, 1)).reduceByKey(_ + _)
})
resultDStream.print
sctx.start
sctx.awaitTermination()
}
}2、有状态转换
（1）updateStateByKey
updateStateByKey操作允许在使用新信息不断更新状态的同时能够保留他的状态
- 定义状态. 状态可以是任意数据类型
- 定义状态更新函数. 指定一个函数, 这个函数负责使用以前的状态和新值来更新状态.
在每个阶段Spark 都会在所有已经存在的key上使用状态更新函数，而不管是否有新的数据在
def updateStateByKey[S: ClassTag](
updateFunc: (Seq[V], Option[S]) => Option[S]): DStream[(K, S)]wordcount案例
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
object StreamingWordCount2 {
def main(args: Array[String]): Unit = {
// 设置将来访问 hdfs 的使用的用户名, 否则会出现全选不够
System.setProperty("HADOOP_USER_NAME", "duoduo")
val conf = new SparkConf().setAppName("StreamingWordCount2").setMaster("local[*]")
// 1. 创建SparkStreaming的入口对象: StreamingContext 参数2: 表示事件间隔
val ssc = new StreamingContext(conf, Seconds(5))
// 2. 创建一个DStream
val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop202", 9999)
// 3. 一个个的单词
val words: DStream[String]=lines.flatMap(_.split("""\s+"""))
// 4. 单词形成元组
val wordAndOne: DStream[(String, Int)] = words.map((_, 1))
// 开始
/*
1. 定义状态: 每个单词的个数就是我们需要更新的状态
2. 状态更新函数. 每个key(word)上使用一次更新新函数
参数1: 在当前阶段 一个新的key对应的value组成的序列 在我们这个案例中是: 1,1,1,1...
参数2: 上一个阶段 这个key对应的value
*/
def updateFunction(newValue: Seq[Int], runningCount: Option[Int]): Option[Int] = {
// 新的总数和状态进行求和操作
newValue.foldLeft(0)(_+_)
val newCount: Int = (0 /: newValue) (_ + _) + runningCount.getOrElse(0)
Some(newCount)
}
// 设置检查点: 使用updateStateByKey必须设置检查点
ssc.sparkContext.setCheckpointDir("ck")
val stateDS: DStream[(String, Int)] = wordAndOne.updateStateByKey[Int](updateFunction _)
//结束
//6. 显示
stateDS.print
//7. 启动流失任务开始计算
ssc.start()
//8. 等待计算结束才推出主程序
ssc.awaitTermination()
ssc.stop(false)
}
}（2）window操作
Spark Streaming 提供了窗口计算，允许执行转换操作作用在一个窗口内的数据
默认情况下，计算只对一个时间段内的RDD进行，有了窗口之后，可以把计算应用到一个指定的窗口内的所有RDD上
一个窗口可以包含多个时间段，基于窗口的操作会在一个比StreamingContext的批次间隔更长的时间范围内，通过整合多个批次的结果，计算出整个窗口的结果
观察上图, 窗口在 DStream 上每滑动一次, 落在窗口内的那些 RDD会结合在一起, 然后在上面操作产生新的 RDD, 组成了 window DStream.
在上面图的情况下, 操作会至少应用在 3 个数据单元上, 每次滑动 2 个时间单位. 所以, 窗口操作需要 2 个参数:
- 窗口长度 – 窗口的持久时间(执行一次持续多少个时间单位)(图中是 3)
- 滑动步长 – 窗口操作被执行的间隔(每多少个时间单位执行一次).(图中是 2 )
注意: 这两个参数必须是源 DStream 的 interval 的倍数
1）reduceByKeyAndWindow(reduceFunc:(V, V) => V, windowDuration: Duration)
val wordAndOne: DStream[(String, Int)] = words.map((_, 1)) /* 参数1: reduce 计算规则 参数2: 窗口长度 参数3: 窗口滑动步长. 每隔这么长时间计算一次. */ val count: DStream[(String, Int)] = wordAndOne.reduceByKeyAndWindow((x: Int, y: Int) => x + y,Seconds(15), Seconds(10))
2）reduceByKeyAndWindow(reduceFunc: (V, V) => V, invReduceFunc: (V, V) => V, windowDuration: Duration, slideDuration: Duration)
比没有invReduceFunc高效.会利用旧值来进行计算.
invReduceFunc: (V, V) => V 窗口移动了, 上一个窗口和新的窗口会有重叠部分, 重叠部分的值可以不用重复计算了.第一个参数就是新的值, 第二个参数是旧的值.
ssc.sparkContext.setCheckpointDir("hdfs://hadoop201:9000/checkpoint")
val count: DStream[(String, Int)] =
wordAndOne
.reduceByKeyAndWindow((x: Int, y: Int) => x + y,
(x: Int, y: Int) => x - y,
Seconds(15),
Seconds(10))3）window(windowLength, slideInterval)
基于对源 DStream 窗化的批次进行计算返回一个新的 Dstream
4）countByWindow(windowLength, slideInterval)
返回一个滑动窗口计数流中的元素的个数
5）countByValueAndWindow(windowLength, slideInterval, [numTasks])
对(K,V)对的DStream调用，返回(K,Long)对的新DStream，其中每个key的的对象的v是其在滑动窗口中频率。如上，可配置reduce任务数量。
