SparkStreaming简单总结一下和几个小案例，整合kafka

我会带着你远行 2023-06-15 07:14 19阅读 0赞

# 1. SparkStreaming是什么 #

> SparkStreaming对于Spark核心API的拓展，从而支持对于实时数据流的可拓展，高吞吐量和容错性流处理。数据可以由多个源取得，例如：Kafka，Flume，Twitter，ZeroMQ，Kinesis或者TCP接口，同时可以使用由如map，reduce，join和window这样的高层接口描述的复杂算法进行处理。最终，处理过的数据可以被推送到文件系统，数据库和HDFS。

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

在内部，其按如下方式运行。Spark Streaming接收到实时数据流同时将其划分为分批，**这些数据的分批将会被Spark的引擎所处理从而生成同样按批次形式的最终流。**(所以讲，sparkStreaming的处理方式依然是批，这一点是和Flink不同的)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 1]

# 2. DStream #

DStream：Discretized Stream，离散流，Spark Streaming提供的一种高级抽象，代表了一个持续不断的数据流；  
DStream可以通过输入数据源来创建，比如Kafka、Flume，也可以通过对其他DStream应用高阶函数来创建，比如map、reduce、join、window；

DStream的内部，**其实是一系列持续不断产生的RDD**，RDD是Spark Core的核心抽象，即，不可变的，分布式的数据集；  
DStream中的每个RDD都包含了一个时间段内的数据；  
**对DStream应用的算子，其实在底层会被翻译为对DStream中每个RDD的操作；**

底层的RDD的transformation操作，**还是由Spark Core的计算引擎来实现的**，Spark Streaming对Spark core进行了一层封装，隐藏了细节，然后对开发人员提供了方便易用的高层次API。

# 3. 从端口接收数据，WordCount #

package SparkStreaming
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.{
        DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{
        Seconds, StreamingContext}
    
    object WordCountStreaming {
        
      def main(args: Array[String]): Unit = {
        
        //配置
        val conf  = new SparkConf().setMaster("local[2]").setAppName("WordCountStreaming")
        //实时数据分析环境对象
        //采集周期，以指定的时间为采集周期
        val Scontext = new StreamingContext(conf,Seconds(3))
        //从指定的端口中采集
        val socketDstream: ReceiverInputDStream[String] = Scontext.socketTextStream("192.168.159.200",9999)
        //将采集的数据进行分解，扁平化
        val wordDstream: DStream[String] = socketDstream.flatMap(line => {
        
          line.split(" ")
        })
        //将数据进行结构的转变，元组
        val tupDstream: DStream[(String, Int)] = wordDstream.map(word => {
        
          (word, 1)
        })
        //聚合
        val reduced: DStream[(String, Int)] = tupDstream.reduceByKey(_+_)
        //将结果打印出来
        reduced.print()
        //不要习惯性的加上stop程序,要让main方法一直运行，与采集器绑定在一起
        //启动采集器
        Scontext.start()
        //Driver等待采集器的执行
        Scontext.awaitTermination()
      }
    }

![在这里插入图片描述][20191121101548926.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 2]

# 4. 从kafka接收数据，手动维护偏移量且追加历史数据 #

package day12.demo03_kafka.sample02_directway
    
    import org.apache.kafka.common.serialization.{
        StringDeserializer}
    import org.apache.spark.SparkContext
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.kafka010.{
        CanCommitOffsets, HasOffsetRanges, KafkaUtils}
    import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
    import org.apache.spark.streaming.{
        Seconds, StreamingContext}
    import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
    
    /**
     * @description 手动维护偏移量，kafka历史+实时WordCount
     * @author YDAlex
     * @data 2019/11/15.
     * @version 1.0
     */
    object demo {
        
      def main(args: Array[String]): Unit = {
        
        val spark: SparkSession = SparkSession.builder()
          .appName("MyWC")
          .master("local")
          .getOrCreate
    
        val sc: SparkContext = spark.sparkContext
        val ssc:StreamingContext = new StreamingContext(sc,Seconds(2))
        ssc.checkpoint("file:///G:\\IntelliJ IDEA 2019.2.1\\Workspace\\sparktest\\data\\ck")//设置检查点本地存储，一般是hdfs
        val topics = Array("test")
        val groupID = "group01"
        //以键值对的方式配置文件（注意这是直连方式并非receiver）
        val KafkaParams = Map[String,Object](
          "bootstrap.servers" -> "hadoop:9092",
          "key.deserializer" -> classOf[StringDeserializer],
          "value.deserializer" -> classOf[StringDeserializer],
          "group.id" -> groupID,
          "auto.offset.reset" -> "earliest",
          "enable.auto.commit" -> (false:java.lang.Boolean)
        )
        val streaming = KafkaUtils.createDirectStream[String, String](
          ssc,
          PreferConsistent,
          Subscribe[String, String](topics, KafkaParams)
        )
        streaming
          .flatMap(_.value().split("\\s+"))
          .filter(_.nonEmpty)
          .map((_,1))
          //历史数据追加
          .updateStateByKey((nowBatch: Seq[Int], historyResult: Option[Int]) => Some(nowBatch.sum + historyResult.getOrElse(0)))
          .print(100)
    
        //wordcount.updateStateByKey(Myfun)
         // .print(100)
        streaming.foreachRDD {
         rdd =>
          //获取该RDD对于的偏移量
          val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
          //更新偏移量
          // some time later, after outputs have completed(将偏移量更新【Kafka】)
          streaming.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
        }
    
    
        ssc.start
        ssc.awaitTermination
      }
      //def Myfun = (nowBatch:Seq[Int],history:Option[Int]) => Some(nowBatch.sum + history.getOrElse(0))
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 3]

# 5. Window时间窗口 #

本函数，适用于求历史一定时长的数据留存，比如天猫双十一时，求历史一小时内的交易总额等信息

object SlideWindowFunDemo {
        
      def main(args: Array[String]): Unit = {
        
        //SparkSession
        val spark: SparkSession = SparkSession.builder()
          .appName(SlideWindowFunDemo.getClass.getSimpleName)
          .master("local[*]")
          .getOrCreate()
    
        val sc: SparkContext = spark.sparkContext
    
    
        val ssc: StreamingContext = new StreamingContext(sc, Seconds(3))
    
        //checkpoint
        //ssc.checkpoint("file:///C:\\Users\\Administrator\\IdeaProjects\\spark-streaming-study\\data\\ck")
    
    
        //DStream,迭代计算,并显示内容
        val ds: DStream[(String, Int)] = ssc.socketTextStream("NODE01", 7777)
          .flatMap(_.split("\\s+"))
          .filter(_.nonEmpty)
          .map((_, 1))
          .cache
    
        //注意：
        //参数2：窗口长度
        //参数3：时间间隔，
        //上述两个参数必须是每个批次时间间隔的整数倍
        //否则针对参数2，报错：Exception in thread "main" java.lang.Exception: The window duration of windowed DStream (4000 ms) must be a multiple of the slide duration of parent DStream (3000 ms)
        //否则针对参数3，报错：Exception in thread "main" java.lang.Exception: The slide duration of windowed DStream (3000 ms) must be a multiple of the slide duration of parent DStream (2000 ms)
        ds.reduceByKeyAndWindow((nowValue: Int, nextValue: Int) => nowValue + nextValue,
          Seconds(6), Seconds(3))
          .print(100)
    
        //启动SparkStreaming应用
        ssc.start
    
        //等待结束（必须要添加）
        ssc.awaitTermination
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191121095651620.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191121095816594.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2,size_16,color_FFFFFF,t_70
[20191121101548926.png]: https://img-blog.csdnimg.cn/20191121101548926.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191121101527703.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20191121102519830.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2,size_16,color_FFFFFF,t_70