[spark streaming]状态操作

梦里梦外; 2022-05-13 03:52 256阅读 0赞

[![96][]][96 1] [Seven\_Ki][96 1] 关注

2018.02.23 22:55\* 字数 1545 阅读 452评论 0喜欢 0

> 本文全部手写原创，请勿复制粘贴、转载请注明出处，谢谢配合！

> **Dstream的转化操作分为无状态的(stateless)和有状态的(stateful)**  
> 无状态转化：每个批次处理都**不依赖**于先前批次的数据，如map() filter() reduceByKey()等均属于无状态的  
> 有状态转化：**依赖**之前的批次数据或者中间结果来计算当前批次的数据，包括**updateStatebyKey()**和**window()**

# 有状态转换操作 #

--------------------

有状态操作是跨时间区间跟踪处理数据的操作。依赖于之前批次的数据。

### Window Operations ###

概念

之前我们说过，我们会为程序设置一个批次时间间隔，如Seconds(2)，那么窗口操作，简单理解就是整合多个批次，计算出整个窗口的结果，因此说有状态操作是跨时间区间的。

![webp][]

Window Operations

假设时间间隔为10s，上面这幅图的意思就是，每隔20s(2个批次间隔)，就对前30s(3个批次间隔)的数据进行整合计算\[time1+time2+time3\]。  
由此可知，一般的window操作会涉及两个参数。

*  对多少个批次进行整合？---window length 窗口长度 - 窗口的持续时间（图中是3）。
 *  间隔多久进行整合操作？---sliding interval 滑动间隔 - 执行窗口操作的时间间隔（图中是2）。  
    **【注意：两者都必须是批次间隔的整数倍】**  
    应用： 每次计算都是最近xx s的数据，微博热点就是基于这个特性 --- 最近30分钟内最热门的头条

常用API

*  window(windowLength, slideInterval) // 返回一个新的Dstream来表示窗口操作的结果数据，也就是说这个新的Dstream中的每个RDD都包括了多个批次中的数据
 *  reduceByWindow(func, windowLength, slideInterval) //在整个窗口上执行归约操作
 *  reduceByKeyAndWindow(func, windowLength, slideInterval, \[numTasks\]) // 对key进行归约操作

scala> val ssc = new StreamingContext(sc, Seconds(5))
    ssc: org.apache.spark.streaming.StreamingContext = org.apache.spark.streaming.StreamingContext@3f908a10
    
    scala> val lines = ssc.socketTextStream("localhost", 7788)
    lines: org.apache.spark.streaming.dstream.ReceiverInputDStream[String] = org.apache.spark.streaming.dstream.SocketInputDStream@7d619bcb
    
    scala> val pairs = lines.flatMap(_.split(" ")).map(word => (word, 1))
    pairs: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.MappedDStream@184b24d4
    
    scala> val wordCounts = pairs.reduceByKey(_ + _)
    wordCounts: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.ShuffledDStream@4bd1541f
    
    scala> wordCounts.print()
    
    scala> val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(15), Seconds(5))
    windowedWordCounts: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.ShuffledDStream@752957da
    
    scala> windowedWordCounts.print()
    
    scala> ssc.start()
    scala> ssc.awaitTermination()

*  reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, \[numTasks\]) //这是上个函数更为高效的版本，需提供一个逆函数invFunc，如+的逆函数是-。  
    为什么更加高效呢？回到上面那张Window Operations图片：  
    window1=time1+time2+time3  
    window2=time3+time4+time5  
    time3重复计算了，影响效率！  
    那么此种方法，就是避免了这些重复计算，它只考虑新进来的和离开的，不考虑之前已经计算过的。  
    window1=time1+time2+time3 => time3=window1-time1-time2  
    window2=window1-time1-time2+time4+time5  
    *+*是对新产生的时间分片（time4,time5内RDD）进行统计，而*-*是对上一个窗口中，过时的时间分片(time1,time2) 进行统计，这样我们就直接利用了上个窗口的计算结果而不需要重新计算。  
    `val windowedWordCounts = pairs.reduceByKeyAndWindow( _+_, _-_, Seconds(15), Seconds(5))`
 *  countByWindow(windowLength, slideInterval) //返回每个窗口中元素个数的Dstream
 *  countByValueAndWindow(windowLength, slideInterval, \[numTasks\])//返回每个窗口中值的个数的Dstream

UpdateStateByKey Operation

该updateStateByKey操作允许你保持任意状态，同时不断用新信息更新它。  
之前我们写过的wordcount程序，我们应该发现了：单词数统计是按批次的，批次与批次互不影响，当你不断输入某个单词的时候，总的计数不会累加。那么如果想要在任意状况下都去更新某个信息，就需要使用UpdateStateByKey ，使用此功能必须执行两个步骤。

定义状态 - 状态可以是任意数据类型。  
定义状态更新函数 - 使用函数指定如何更新状态。

> def **updateStateByKey**\[S\](updateFunc: (Iterator\[(K, Seq\[V\], Option\[S\])\]) ? Iterator\[(K, S)\], partitioner: Partitioner, rememberPartitioner: Boolean)(implicit arg0: ClassTag\[S\]): DStream\[(K, S)\]  
> Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values of each key. org.apache.spark.Partitioner is used to control the partitioning of each RDD.
> 
>  *  S：State type
>  *  updateFunc：State update function. Note, that this function may generate a different tuple with a different key than the input key. Therefore keys may be removed or added in this way. It is up to the developer to decide whether to remember the partitioner despite the key being changed.
>  *  partitioner：Partitioner for controlling the partitioning of each RDD in the new DStream
>  *  rememberPartitioner：Whether to remember the partitioner object in the generated RDDs.

scala> val ssc = new StreamingContext(sc, Seconds(5))
    ssc: org.apache.spark.streaming.StreamingContext = org.apache.spark.streaming.StreamingContext@3f908a10
    
    scala> val lines = ssc.socketTextStream("localhost", 9999)
    lines: org.apache.spark.streaming.dstream.ReceiverInputDStream[String] = org.apache.spark.streaming.dstream.SocketInputDStream@f500481
    
    // * String : 单词  * Seq[Int] ：单词在当前批次出现的次数  * Option[Int] ： 历史批次出现的次数
    scala> val updateFunc = (iter: Iterator[(String, Seq[Int], Option[Int])]) => {iter.flatMap{case(x,y,z)=>Some(y.sum + z.getOrElse(0)).map(m=>(x, m))}}
    updateFunc: Iterator[(String, Seq[Int], Option[Int])] => Iterator[(String, Int)] = <function1>
    
    scala> val results = lines.flatMap(_.split(" ")).map((_,1)).updateStateByKey(updateFunc, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)
    results: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.StateDStream@51e0629a
    
    scala> results.print()
    
    scala> ssc.checkpoint("hdfs://master/user/checkpoint")//必须设置检查点机制 不然会报错
    
    scala> ssc.start()
    
    scala> ssc.awaitTermination()
    -------------------------------------------
    Time: 1519396875000 ms
    -------------------------------------------

UpdateStateByKey用处：统计广告点击流量，统计这一天的车流量。

【注意：某些有状态操作（UpdateStateByKey、reduceByKeyAndWindow）需要使用检查点机制，将在后面详细介绍】

# 无状态转化操作 #

--------------------

和RDD的转化操作很类似，DStreams支持Spark RDD中的许多转换

*  map(func)
 *  flatMap(func)
 *  filter(func)
 *  count()
 *  union(otherStream)
 *  reduce(func)
 *  reduceByKey(func, \[numTasks\])
 *  repartition(numPartitions)： 改变Dstream分区数  
    以上较简单，请参考本人其他博客：[Spark入门教程（五）][Spark]或者[DStream API文档][DStream API]

*  **transform(func)** Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream.  
    这个函数很牛逼，它可以让你直接操作其内部的RDD，也就是说，如果这些无状态操作都不够用的时候，你想要的东西在API中却没有的时候，你可以自己提供任意一个RDD到RDD的函数，这个函数在数据流每个批次中都被调用，生成一个新的流。  
    transform常见应用就是让你重用你之前为RDD写的批处理代码。
 *  **join(otherStream, \[numTasks\])**  
    无状态转换操作也可以在多个Dstream中整合数据，比如使用RDD中讲解过的join方法。

Stream-stream joins

val stream1: DStream[String, String] = ...
    val stream2: DStream[String, String] = ...
    val joinedStream = stream1.join(stream2)

![webp 1][]

join操作演示.gif

Stream-dataset joins  
join方法作用是两个DStream，如果参数不是DStream，那就使用上面的transform 方法

val dataset: RDD[String, String] = ...
    val windowedStream = stream.window(Seconds(20))...
    val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }

**最后需要注意的是**： 我们知道Dstream是一系列的RDD序列流，由许多RDD（批次）组成。无状态转化操作是**分别应用在每个RDD（批次）上**的，例如，reduceByKey会归约每个时间区间的数据，但不同区间数据是不会归约的。

[96]: /images/20220513/fb3eb39fbdbb4639b8acdc68d963163b.png
[96 1]: https://www.jianshu.com/u/bdda6502e5a2
[webp]: /images/20220513/a8a25f25e9164eb0bf097967e0be9ae8.png
[Spark]: https://www.jianshu.com/p/f3dd1bb19586
[DStream API]: https://link.jianshu.com/?t=http%3A%2F%2Fspark.apache.org%2Fdocs%2F2.2.0%2Fapi%2Fscala%2Findex.html%23org.apache.spark.streaming.dstream.DStream
[webp 1]: /images/20220513/11c5b9f767be47fabc7119a510cdefbd.png