有興趣想學習國內整套Spark+Spark Streaming+Machine learning頂級課程的，可加我qq  471186150。共享視訊，價效比超高！

1：SparkStreaming雖然說已經支援了很多不同型別的資料來源。但是有時候可能我們的一些資料來源非常特殊 ，不是它天然預設支援的，這時候就要自定義Receiver。而自定義Receiver，一般都是基於網路的方式。因為你傳資料的話，一般是從另外一個網路埠傳過來，至於傳的協議是另外一碼事。

2：從本質上來說，SparkStreaming中的所有Receiver，都是自定義的Receiver。所以你要想自定義一個Receiver，最最簡單的方式，你就是看下已有的Receiver怎麼去實現。

具體步驟：http://spark.apache.org/docs/latest/streaming-custom-receivers.html

class CustomReceiver(host: String, port: Int)
  extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) with Logging {

  def onStart() {
    // Start the thread that receives data over a connection
    new Thread("Socket Receiver"
 
) {
      override def run() { receive() }//開啟執行緒調receiver()方法
    }.start()
  }

  def onStop() {
   // There is nothing much to do as the thread calling receive()
   // is designed to stop by itself if isStopped() returns false
  }

  /** Create a socket connection and receive data until receiver is stopped */
 

  private def receive() {
    var socket: Socket = null
    var userInput: String = null
    try {
     // Connect to host:port，Receiver的時候就連上Socket
     socket = new Socket(host, port)

     // Until stopped or connection broken continue reading
     val reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), "UTF-8"))
     userInput = reader.readLine()
     while(!isStopped && userInput != null) {
       store(userInput)
       userInput = reader.readLine()//每讀一行，存一次，一直迴圈
     }
     reader.close()
     socket.close()

     // Restart in an attempt to connect again when server is active again
     restart("Trying to connect again")//先stop，然後再start()
    } catch {
     case e: java.net.ConnectException =>
       // restart if could not connect to server
       restart("Error connecting to " + host + ":" + port, e)
     case t: Throwable =>
       // restart if there is any other error
       restart("Error receiving data", t)
    }
  }
}

// Assuming ssc is the StreamingContext
val customReceiverStream = ssc.receiverStream(new CustomReceiver(host, port))
val words = lines.flatMap(_.split(" "))
...