SparkStreaming-有状态的数据转换

偏执的太偏执、 2023-02-19 13:29 21阅读 0赞

DStream上的原语与RDD的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种Window相关的原语。  
**`无状态转化操作`**  
无状态转化操作就是把简单的RDD转化操作应用到每个批次上，也就是转化DStream中的每一个RDD。部分无状态转化操作列在了下表中。注意，针对键值对的DStream转化操作(比如 reduceByKey())要添加import StreamingContext.\_才能在Scala中使用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_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()来合并多个流。  
**`有状态转化操作`**  
**`UpdateStateByKey`**  
UpdateStateByKey原语用于记录历史记录，有时，我们需要在 DStream 中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey() 为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。给定一个由(键，事件)对构成的 DStream，并传递一个指定如何根据新的事件 更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为(键，状态) 对。  
updateStateByKey() 的结果会是一个新的 DStream，其内部的 RDD 序列是由每个时间区间对应的(键，状态)对组成的。  
updateStateByKey操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，你需要做下面两步：

1.  定义状态，状态可以是一个任意的数据类型。
2.  定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。  
    **`使用updateStateByKey需要对检查点目录进行配置，会使用检查点来保存状态。`**  
    需要在kafka创建

bin/kafka-topics.sh --zookeeper hadoop102:2181 --create --topic donglin --partitions 3 --replication-factor 2

执行命令

bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic donglin

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    
    object SparkStream_kafka { 
    
      def main(args: Array[String]): Unit = { 
    
        //使用SparkStream完成WordCount
    
        //Spark配置对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
    
        //实时数据分析的环境对象
        //采集周期，以指定的时间为周期采集实时数据
        val streamingContext = new StreamingContext(sparkConf, Seconds(5))
    
        //从kafka中采集数据
        val kafkaDStream = KafkaUtils.createStream(
          streamingContext,
          "hadoop102:2181",
          "donglin",
          Map("donglin" -> 3)
        )
    
        //将采集的数据进行分解（扁平化）
        val wordDStream = kafkaDStream.flatMap(t => t._2.split(" "))
    
        //将数据进行结构的转换，方便统计分析
        val mapDStream = wordDStream.map((_, 1))
    
        //将转换结构的数据进行聚合处理
      val wordToSumDStream = mapDStream.reduceByKey(_ + _)
    
    
        //将结果打印出来
        wordToSumDStream.print()
    
        //能停止采集程序
    
        //启动采集器
        streamingContext.start()
        //Driver等待采集器执行
        streamingContext.awaitTermination()
    
      }
    
    }

窗口函数  
**`window`**  
代码如下

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.kafka.KafkaUtils
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    
    object SparkStream_window { 
    
      def main(args: Array[String]): Unit = { 
    
        //使用SparkStreaming的窗口
    
        //Spark配置对象
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
    
        //实时数据分析的环境对象
        //采集周期，以指定的时间为周期采集实时数据
        val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    
        //从kafka中采集数据
        val kafkaDStream = KafkaUtils.createStream(
          streamingContext,
          "hadoop102:2181",
          "donglin",
          Map("donglin" -> 3)
        )
    
        //窗口大小应该为采集周期的整数倍,窗口华东的步长应该为采集周期的整数倍
        val windowDStream = kafkaDStream.window(Seconds(9), Seconds(3))
    
        //将采集的数据进行分解（扁平化）
        val wordDStream = windowDStream.flatMap(t => t._2.split(" "))
    
        //将数据进行结构的转换，方便统计分析
        val mapDStream = wordDStream.map((_, 1))
    
        //将转换结构的数据进行聚合处理
      val wordToSumDStream = mapDStream.reduceByKey(_ + _)
    
    
        //将结果打印出来
        wordToSumDStream.print()
    
        //能停止采集程序
    
        //启动采集器
        streamingContext.start()
        //Driver等待采集器执行
        streamingContext.awaitTermination()
      }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200620233002365.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70