自定义SparkStreaming的接收器接收不同类型的数据源

╰半夏微凉° 2023-01-12 10:14 100阅读 0赞

# 自定义SparkStreaming的接收器接收不同类型的数据源 #

### 步骤： ###

1、自定义一个类，继承Receiver

abstract  class  Receiver[T](val  storeLevel: StorageLevel)  extends Serializable
         [T]: Receiver中接收到的数据的类型
         StoreLevel：存储级别，Reciver接收的数据怎么存，存哪里

2、实现两个抽象方法

onStart(): 必须实现的方法！
                 会在Receiver开始接收数据前，执行一些必要的安装步骤，例如开启线程，开启socket，开启流等！！！
                 onStart（）线程不能被阻塞，需要另起一个线程接收数据！！
                 接收到的数据，调用store 方法，将数据保存到指定的地方
                 
     onStop(): 必须实现的方法
                 会在Receiver停止接收数据前，执行一些必要的清理步骤，清理掉已经安装的流，socket，停止线程等操作。

3、异常处理：

*  要么重启receiver，调用restart()
 *  要么彻底停止，调用stop()

### 代码实例 ###

package com.saprkstreaming
    
    import java.io.{ BufferedReader, InputStreamReader}
    import java.net.Socket
    
    import org.apache.spark.storage.StorageLevel
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.DStream
    import org.apache.spark.streaming.receiver.Receiver
    
    
    /* 自定义接收器从一个端口中接收文本数据！ */
    
    object NewReceiver { 
      
      def main(args: Array[String]): Unit = { 
      
        //创建StreamingContext，每3s采集一批数据
         val context: StreamingContext = new StreamingContext("local[*]","sc",Seconds(3))
      
        //使用自己定义的Receiver接收器，从端口获取文本数据
        val receiver: MyReceiver = new MyReceiver("hadoop102",3333)
        
        
        //从自定义的接收器中获取一个DStream
        val ds = context.receiverStream(receiver)
      
        //对DStream中数据进行转换
        val result: DStream[(String, Int)] = ds.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)
      
        //输出结果，可以保存到文件，也可以在屏幕打印
        result.print(100)
      
        //启动应用
        context.start()
      
        //阻塞当前进程，防止运行结束关闭
        //一直阻塞当前进程，直到app手动调用了stop或者由于异常终止
        context.awaitTermination()
      
      }
      
    }
    
    //自定义的Receiver类
    class MyReceiver(var hostname:String, var port:Int)  extends Receiver[String](StorageLevel.MEMORY_ONLY){ 
      
      var socket:Socket=_
      var reader:BufferedReader =_
      
      def receiveData() = { 
      
        //启动一个新的新城接受数据
        new Thread{ 
        
          //设置为守护线程
          setDaemon(true)
        
          //接收数据
          override def run() = { 
          
            //返回一行的内容，当读到流的末尾返回null
            //读到一行数据就返回，否则就一直阻塞
            var line: String = reader.readLine()
          
            try { 
              while (socket.isConnected && line != null) { 
                //读到数据后，将数据进行存储
                store(line)
                //继续读取数据
                line = reader.readLine()
              }
            } catch { 
              case e =>
            } finally { 
              restart("重启继续尝试读取数据！")
              onStop()
            }
          }
        }.start()
      }
      
      
      /** * 从一个端口中接收文本数据： * ①创建一个Socket连接到指定的端口 * ②从Socket中获取数据流，从流中读取数据 */
      override def onStart(): Unit = { 
        
        try { 
          socket = new Socket(hostname, port)
        } catch { 
          case e:Exception =>{ 
            onStop()
            restart("socket连接异常，重启尝试！")
            return
          }
        }
        
        println(hostname+":"+port+"已经连接上！")
        
        //创建BufferReader流，可以一行一行的读取文本数据
        reader = new BufferedReader(new InputStreamReader(socket.getInputStream,"utf-8"))
        
        //新启动一个线程接收数据
        receiveData()
        
      }
      
      
      //关闭相关资源的方法
      override def onStop(): Unit = { 
        
        if (reader != null){ 
          reader.close()
          reader=null
        }
        
        if (socket!=null){ 
          socket.close()
          socket = null
        }
        
      }
    }

### 相关依赖 ###

<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>