SparkStreaming编程与Queue测试流

青旅半醒 2023-01-11 15:27 132阅读 0赞

## 一、StreamingContext如何创建？如何使用？ ##

创建：  ①基于SparkConf和Duration
                ②基于SparkContext和Duration
                ③基于masterurl，appname，和Duration
                
        使用：  ①构造StreamingContext对象
                ②根据数据源，构造DStream
                ③对DStream进行各种转换(运算)
                ④开启app，执行运算  StreamingContext.start()
                ⑤流式应用，因此启动后，需要一直运行，阻塞当前driver的启动线程，
                        StreamingContext.awaitTermination
                ⑥在需要关闭时，
                           StreamingContext.stop()

## 二、pom依赖 ##

<dependencies>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.12</artifactId>
                <version>3.0.0</version>
            </dependency>
    
        </dependencies>

## 三、代码 ##

package com.saprkstreaming
    
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object WordCount { 
       
      /* StreamingContext： ① 使用SparkStreaming为核心入口，用来从多种不同数据源构造DStream Duration：应用每批数据的采集间隔，采集周期 Milliseconds Seconds Minutes 使用上述的apply方法构造 */
        def main(args: Array[String]): Unit = { 
      
          //创建StreamingContext，每3s采集一批数据
          val context: StreamingContext = new StreamingContext("local[*]","sc",Seconds(3))
          
          //执行DStream的构造（根据数据来源构造）和转换
          //如果调用的是rdd中的方法，称为算子（转换算子，行动算子）
          //如果调用DStream中的方法，称为抽象原语
          val ds = context.socketTextStream("hadoop102",3333)
        
          //对DStream中数据进行转换
          val result: DStream[(String, Int)] = ds.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)
      
          //输出结果，可以保存到文件，也可以在屏幕打印
          result.print(100)
      
          //启动应用
          context.start()
          
          //阻塞当前进程，防止运行结束关闭
          //一直阻塞当前进程，直到app手动调用了stop或者由于异常终止
          context.awaitTermination()
      
      }
    }

## 三、测试 ##

在linux中开放端口进行测试

[hadoop@hadoop102 ~]$ nc -lk hadoop102 3333
    a b c d
    h l

输出结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

## 四、创建QueueStream流进行逻辑测试 ##

package com.saprkstreaming
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import scala.collection.mutable
    
    /* 从队列中获取DS，主要用于测试程序的逻辑是否有问题！！！ def queueStream[T: ClassTag]( queue: Queue[RDD[T]], //Queue中必须存放的是一个个RDD oneAtATime: Boolean = true //每个采集周期是否只从Queue中消费一个RDD ): InputDStream[T] */
    
    object QueueExec { 
      
      def main(args: Array[String]): Unit = { 
      
        //创建StreamingContext，每3s采集一批数据
        val context: StreamingContext = new StreamingContext("local[*]","sc",Seconds(3))
        
        //将源源不断的数据存放到RDD中，模拟从queue中获取流式数据
        val queue= mutable.Queue[RDD[String]]()
        
        //获取一个queueStream流
        val ds: InputDStream[String] = context.queueStream(queue,false)
       
        //对DStream中数据进行转换
        val result: DStream[(String, Int)] = ds.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)
      
        //输出结果，可以保存到文件，也可以在屏幕打印
        result.print(100)
      
        //启动应用
        context.start()
        
        
        val list: List[String] = List("a b c","b c d")
        
        //获取当前应用的SparkContext
        val rdd: RDD[String] = context.sparkContext.makeRDD(list)
        
        //模拟数据源源不断的产生
        for (i <- 1 to 1000){ 
          queue.enqueue(rdd)
          Thread.sleep(1000)
        }
        
      
        //阻塞当前进程，防止运行结束关闭
        //一直阻塞当前进程，直到app手动调用了stop或者由于异常终止
        context.awaitTermination()
        
      } 
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221119/27328ac1efff45838afbff943dcd3814.png