Spark无状态和有状态转化操作介绍

怼烎@ 2023-06-15 09:45 15阅读 0赞

## 1. 无状态转化操作 ##

无状态转化操作就是把简单的`RDD`转化操作应用到每个批次上，也就是转化`DStream`中的每一个`RDD`。部分无状态转化操作列在了下表中。注意，**针对键值对的DStream转化操作(比如 reduceByKey())要添加import StreamingContext.\_才能在Scala中使用**。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]  
需要记住的是，尽管这些函数看起来像作用在整个流上一样，**但事实上每个DStream在内部是由许多RDD(批次)组成，且无状态转化操作是分别应用到每个RDD上的**。例如，reduceByKey()会归约每个时间区间中的数据，但不会归约不同区间之间的数据。

举个例子，在之前的wordcount程序中，我们只会统计5秒内接收到的数据的单词个数，而不会累加。

无状态转化操作也能在多个DStream间整合数据，**不过也是在各个时间区间内**。例如，键 值对DStream拥有和RDD一样的与连接相关的转化操作，也就是`cogroup()、join()、leftOuterJoin()` 等。我们可以在DStream上使用这些操作，这样就对每个批次分别执行了对应的RDD操作。  
我们还可以像在常规的Spark 中一样使用 DStream的union() 操作将它和另一个DStream 的内容合并起来，也可以使用StreamingContext.union()来合并多个流。

## 2. 有状态转化操作： ##

### 2.1 DStream有状态操作之`updateStateByKey`操作 ###

**需求:实现跨批次之间维护状态（例如流计算中累加`wordcount`）**

在[https://blog.csdn.net/qq\_35885488/article/details/102667468][https_blog.csdn.net_qq_35885488_article_details_102667468]基础上操作

**先上代码**

object UpdateStateByKey {
      def main(args: Array[String]): Unit = {
        System.setProperty("HADOOP_USER_NAME", "hadoop")
        var sparkConf =new SparkConf().setMaster("local[*]").setAppName("UpdateStateByKey").set("spark.testing.memory", "2147480000")
        
       //分析环境对象以及采集周期
        val streamContext = new StreamingContext(sparkConf,Seconds(10))
        //保存数据的状态，设置检查点路径
        streamContext.checkpoint("hdfs://192.168.181.110:9000/spark-streaming")
        //设置socket源
        val socketStreamLine: ReceiverInputDStream[String] = streamContext.socketTextStream("192.168.181.110",9999)
    	
    	//进行dstream操作，updateStateByKey实现了跨批次状态维护
        val print1: Unit = socketStreamLine.flatMap(_.split(" ")).map((_, 1))
          .updateStateByKey[Int](updateFunction _).print()
    
        //启动采集器
        streamContext.start()
        //等待采集器执行
        streamContext.awaitTermination()
    
      }
    
      /**
        *
        * @param currentValues
        * @param preValues
        * @return
        */
        //currentValues   preValues两个输入参数的数据来源是updateStateByKey[Int]
      def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
        val current = currentValues.sum
        val pre = preValues.getOrElse(0)
        Some(current + pre)
      }
    
    }

**代码核心1**

//定义状态更新函数
    def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
        val current = currentValues.sum
        val pre = preValues.getOrElse(0)
    
        Some(current + pre)
      }

**核心二**

//设置检查点
        // 如果使用了stateful的算子，必须要设置checkpoint
        // 在生产环境中，建议大家把checkpoint设置到HDFS的某个文件夹中
        ssc.checkpoint("hdfs://192.168.181.110:9000/spark-streaming")

进行词频统计时，首先`updateStateByKey`操作会将当前批次到达的词频进行`key`相同的进行归并。

注意：设置检查点可能要修改文件夹权限`hadoop fs -chmod -R 777 /dic`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]  
**归并的规则**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]

//currentValues: Seq[Int],====>(hadoop(1,1,1,1))，（spark(1,2,1)）
    // preValues: Option[Int] 维护当前历史状态的词频
    
    updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int]

val current = currentValues.sum  当前的数据求和操作
    
     val pre = preValues.getOrElse(0)   取历史，如果为空赋值为0

**这里就不贴结果了，肯定能跑起来的…**

完整代码托管于`github`地址：[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191121163024672.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_qq_35885488_article_details_102667468]: https://blog.csdn.net/qq_35885488/article/details/102667468
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2019112116393059.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191121164132631.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1ODg1NDg4,size_16,color_FFFFFF,t_70
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark