Spark Streaming快速入门系列(4) | 无状态和有状态操作

╰+攻爆jí腚メ 2022-12-04 02:29 358阅读 0赞

### 目录 ###

*  一.无状态转换操作
 *  二.有状态转换操作
 *   *  1.updateStateByKey
     *  2.window
     *  window的基本用法
     *  window的优化操作
     *  window函数

# 一.无状态转换操作 #

![在这里插入图片描述][20200905155625212.png_pic_center]

官网  
[http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html\#transformations-on-dstreams][http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html_transformations-on-dstreams]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

*  transform 原语允许 DStream上执行任意的RDD-to-RDD函数。
 *  可以用来执行一些 RDD 操作, 即使这些操作并没有在 SparkStreaming 中暴露出来.  
    该函数每一批次调度一次。其实也就是对DStream中的RDD应用转换。
 *  代码实现

import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.SparkConf
    
    object Transform { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
        val sourceStream = ssc.socketTextStream("hadoop102", 9999)
        val result = sourceStream.transform(rdd => { 
           rdd.flatMap(_.split("\\W+")).map((_,1)).reduceByKey(_+_)
        })
        result.print()
        
        ssc.start()
        ssc.awaitTermination()
      }
    }

# 二.有状态转换操作 #

## 1.updateStateByKey ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]

代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object UpstateByKeyDemo1 { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        ssc.checkpoint("ck1")
    
        val sourceStream = ssc.socketTextStream("hadoop102", 9999)
        sourceStream.flatMap(_.split("\\W+"))
            .map((_,1))
            .updateStateByKey((seq: Seq[Int], opt: Option[Int])=>{ 
              Some(seq.sum + opt.getOrElse(0))
            })
            .print(1000)
    
        ssc.start()
        ssc.awaitTermination()
    
      }
    
    }

结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]

## 2.window ##

## window的基本用法 ##

*  窗口操作需要 2 个参数:

1.  窗口长度 – 窗口的持久时间(执行一次持续多少个时间单位)
2.  滑动步长 – 窗口操作被执行的间隔(每多少个时间单位执行一次).  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]

代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Window1 { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        ssc
          .socketTextStream("hadoop102",9999)
          .flatMap(_.split("\\W+"))
          .map((_,1))
          .reduceByKeyAndWindow(_+_,Seconds(9),slideDuration = Seconds(6))
            .print()
    
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

## window的优化操作 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]  
代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Window2 { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
        //优化时要加checkpoint
        ssc.checkpoint("ck1")
    
        ssc
          .socketTextStream("hadoop102",9999)
          .flatMap(_.split("\\W+"))
          .map((_,1))
          .reduceByKeyAndWindow(_+_,_-_,Seconds(9),slideDuration = Seconds(6),filterFunc = _._2>0)
            .print()
    
        ssc.start()
        ssc.awaitTermination()
      }
    }

解释说明(filterFunc = \_.\_2>0)  
结果会是这样  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]  
**`那零怎么来的！`**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]

## window函数 ##

代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object Window3 { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        ssc
          .socketTextStream("hadoop102",9999)
          .window(Seconds(9),Seconds(6))
          .flatMap(_.split("\\W+"))
          .map((_,1))
          .reduceByKey(_+_)
            .print()
    
        ssc.start()
        ssc.awaitTermination()
      }
    
    }

[20200905155625212.png_pic_center]: /images/20221123/0bc4480081e2455b84f248830c88e45c.png
[http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html_transformations-on-dstreams]: http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html#transformations-on-dstreams
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/2506ea5708034c8ebfd3357a7794c523.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/7a6a96be788c4aac86514ead76126d16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/f1f3869e0acd47fcb924ef0f878c27ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/cb1daabdabaf4b50bc90475281784974.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/7ac5f28190524e12ac757366a68bbd5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/92e81b2bd84547c882f3d74edf0c81bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/a580c7047d5c490191fa027c984f7716.png