Spark Streaming快速入门系列(2) | RDD队列+自定义数据源

亦凉 2022-12-04 02:24 162阅读 0赞

### 目录 ###

*  一.RDD队列
 *  二.自定义数据源

# 一.RDD队列 #

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

*  1.用法及说明  
    测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理。
 *  2.案例实操  
    需求：循环创建几个 RDD，将 RDD 放入队列。通过 Spark Streaming创建 Dstream，计算 WordCount
 *  3.代码实现

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object WordCount1 { 
      def main(args: Array[String]): Unit = { 
        //1.创建streamingContext
        val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount1")
        val ssc = new StreamingContext(conf, Seconds(3))
        //2.从数据源创建一个流：socket rdd队列 自定义接收器， kafka(重点)
        val sourceStream = ssc.socketTextStream("hadoop102",9999)
        //3.对流做各种转换
        val result = sourceStream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
        //4.行动算子：print foreach foreachRDD
        result.print()    //把结果打印在控制台
        //5.启动流
        ssc.start()
        //6.阻止主线程退出
        ssc.awaitTermination()
    
      }
    
    }

# 二.自定义数据源 #

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

*  1.使用及说明  
    其实就是自定义接收器  
    需要继承Receiver，并实现onStart、onStop方法来自定义数据源采集。
 *  2.案例实操  
    需求:  
    自定义数据源，实现监控某个端口号，获取该端口号内容。
 *  3.代码实现

import java.io.{ BufferedReader, InputStreamReader}
    import java.net.Socket
    
    import org.apache.spark.SparkConf
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.receiver.Receiver
    
    object MyReceiverDemo { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount2")
        val ssc = new StreamingContext(conf, Seconds(3))
    
        val sourceStream = ssc.receiverStream(new MyReceiver("hadoop102", 9999))
    
        sourceStream.flatMap(_.split(" "))
          .map((_,1))
          .reduceByKey(_+_)
          .print()
    
        ssc.start()
        ssc.awaitTermination()
    
      }
    
    }
    // 接收器从socket接收数据
    class MyReceiver(host :String,port:Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){ 
    
      var socket:Socket = _
      var reader:BufferedReader = _
      override def onStart(): Unit = { 
        //连接socket 去读取数据 socket 编程
        runInThread{ 
          try{ 
            val socket = new Socket(host, port)
            val reader = new BufferedReader(new InputStreamReader(socket.getInputStream, "utf-8"))
            var line = reader.readLine()
            //当对方发送一个流结束标志的时候，会受到null
            while (line != null && socket.isConnected){ 
              store(line)
              line = reader.readLine()   //如果流中没有数据，这里会一直堵塞
            }
          }catch { 
             //case e => println(e.getMessage)
            case e => e.printStackTrace()
          }finally { 
            restart("重启接收器")
            //立即调用onStop方法，再调用onStart方法
          }
        }
      }
    
      //在一个子线程中执行传入的代码
      def runInThread(op: =>Unit)={ 
         new Thread(){ 
           override def run(): Unit = op
    
         }.start()
    
      }
    
    
      //释放资源
      override def onStop(): Unit = { 
        if (socket != null) socket.close()
        if (reader != null) reader.close()
      }
    }

*  4.开启端口

nc -lk 10000

[20200905140910240.png_pic_center]: /images/20221123/86e386e14a3a4a45952c6f886bbda1e2.png
[2020090514085648.png_pic_center]: /images/20221123/a4974a90c9084e2599b9166f0ef63e31.png