可以繼承Receiver類來實(shí)現(xiàn)自定義采集器，需要實(shí)現(xiàn)方法如下：

• onStart：接收器啟動(dòng)方法
• onStop：接收器停止方法

同時(shí)還要指定Receiver的緩存等級

1. 代碼實(shí)現(xiàn)

   class MyReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {
   
     var socket: Socket = _
   
     override def onStart(): Unit = {
       new Thread(new Runnable {
         override def run(): Unit = {
           receive()
         }
       }).start()
     }
   
     override def onStop(): Unit = {
       if(socket != null) {
         socket.close()
         socket = null
       }
     }
   
     def receive() {
       socket = new Socket(host, port)
       val reader = new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))
       var line: String = null
       while ((line = reader.readLine()) != null) {
         this.store(line)
       }
     }
   
   }
   

2. 使用自定義接收器

   import org.apache.spark.SparkConf
   import org.apache.spark.storage.StorageLevel
   import org.apache.spark.streaming.{Seconds, StreamingContext}
   
   
   object MyReceiverDemo {
     def main(args: Array[String]): Unit = {
   
       val conf = new SparkConf().setMaster("local[4]").setAppName("MyReceiverDemo")
       val streamingContext = new StreamingContext(conf, Seconds(5))
   
       // 使用StreamingContext對象的receiverStream方法，指定自定義Receiver接收數(shù)據(jù)
       val receiverDStream = streamingContext.receiverStream(new MyReceiver("192.168.0.100", 9999))
   
       val flatMapDStream = receiverDStream.flatMap(_.split(" "))
   
       val mapDStream = flatMapDStream.map((_, 1))
   
       val reduceByKeyDStream = mapDStream.reduceByKey(_ + _)
   
       reduceByKeyDStream.print()
   
       streamingContext.start()
       streamingContext.awaitTermination()
     }
   }