// 聲明采集器
// 1）繼承Receiver
// 2) 重寫方法 onStart，onStop

package date_10_16_SparkStreamingimport java.io.{BufferedReader, InputStreamReader} import java.net.Socketimport org.apache.spark.SparkConf import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.receiver.Receiverobject MyReceiver{def main(args: Array[String]): Unit = {//使用SparkStreaming完成wordcount//配置對象val conf = new SparkConf().setMaster("local[*]").setAppName("wordcount")//實時數據分析的環境對象//StreamingContext需要兩個參數，一個conf，一個是采集周期val streamingContext = new StreamingContext(conf,Seconds(5))//從指定的端口中采集數據val socketLineDstream = streamingContext.receiverStream(new MyReceiver1("chun1",9999))//將采集的數據進行分解(扁平化)val wordToSumDstream = socketLineDstream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)wordToSumDstream.print()//這里不能停止采集功能，也就是streamingContext不能結束//可以簡單理解為啟動采集器streamingContext.start()//Driver等待采集器，采集器不挺Driver不停止streamingContext.awaitTermination()}}// 聲明采集器 // 1）繼承Receiver // 2) 重寫方法 onStart，onStop class MyReceiver1(host:String,port:Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){var socket : Socket = nulldef receive(): Unit = {socket = new Socket(host,port)val reader = new BufferedReader(new InputStreamReader(socket.getInputStream,"UTF-8"))var line : String = nullwhile ((line = reader.readLine()) != null){//將采集器的數據存儲到采集器內部進行轉換if ("END".equals(line)){return}else{this.store(line)}}}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}} }

總結

以上是生活随笔為你收集整理的SparkStreaming - 自定义数据源(自定义采集器)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。