Spark系列--Spark Streaming(三)DStreams输入--基本数据源及自定义Receiver

水深无声 2022-05-18 01:27 448阅读 0赞

## 前言 ##

Spark Streaming原生支持一些不同的数据源。一些“核心”数据源已经被打包到Spark Streaming 的 Maven 工件中，而其他的一些则可以通过 spark-streaming-kafka 等附加工件获取。每个接收器都以 Spark 执行器程序中一个长期运行的任务的形式运行，因此会占据分配给应用的 CPU 核心。此外，我们还需要有可用的 CPU 核心来处理数据。这意味着如果要运行多个接收器，就必须至少有和接收器数目相同的核心数，还要加上用来完成计算所需要的核心数。例如，如果我们想要在流计算应用中运行 10 个接收器，那么至少需要为应用分配 11 个 CPU 核心。所以如果在本地模式运行，不要使用local或者local\[1\]。

**DStreams输入的数据源主要分为两大类：基本数据源和高级数据源。**  
基本数据源主要如下：

*  socket数据流
 *  文件数据源
 *  RDD队列

**另外可以–自定义数据接收器Receiver**

**Socket数据流（socketTextStream）前面已经介绍过，本篇介绍下文件数据源、RDD队列以及如何自定义Receiver。**

## 一、文件数据源 ##

文件数据流：能够读取所有HDFS API兼容的文件系统文件，通过fileStream方法进行读取。

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

Spark Streaming 将会监控 dataDirectory 目录并不断处理移动进来的文件，记住目前不支持嵌套目录。

*  文件需要有相同的数据格式。
 *  文件进入dataDirectory的方式需要通过移动或者重命名来实现。
 *  一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据。
 *  如果文件比较简单，则可以使用 streamingContext.textFileStream(dataDirectory)方法来读取文件。文件流不需要接收器，不需要单独分配CPU核。

**Hdfs读取实例：**

**提前需要在HDFS上建好目录。**

scala> import org.apache.spark.streaming._
    import org.apache.spark.streaming._
    
    scala> val ssc = new StreamingContext(sc, Seconds(1))
    ssc: org.apache.spark.streaming.StreamingContext = org.apache.spark.streaming.StreamingContext@4027edeb
    
    scala> val lines = ssc.textFileStream("hdfs://hadoop0:9000/data/")
    lines: org.apache.spark.streaming.dstream.DStream[String] = org.apache.spark.streaming.dstream.MappedDStream@61d9dd15
    
    scala> val words = lines.flatMap(_.split(" "))
    words: org.apache.spark.streaming.dstream.DStream[String] = org.apache.spark.streaming.dstream.FlatMappedDStream@1e084a26
    
    scala> val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
    wordCounts: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.ShuffledDStream@8947a4b
    
    scala> wordCounts.print()
    
    scala> ssc.start()

**上传文件：**

[root@hadoop0 hadoop]$ ls
    bin  data  etc  include  lib  libexec  LICENSE.txt  logs  NOTICE.txt  README.txt  sbin  sdata  share
    [root@hadoop0 hadoop]$ bin/hdfs dfs -put ./LICENSE.txt /data/
    [root@hadoop0 hadoop]$ bin/hdfs dfs -put ./README.txt /data/

**计算结果如下：**

------------------------------------------- Time: 1504665716000 ms -------------------------------------------
    ------------------------------------------- Time: 1504665717000 ms -------------------------------------------
    ------------------------------------------- Time: 1504665718000 ms -------------------------------------------
    (227.7202-1,2)
    (created,2)
    (offer,8)
    (BUSINESS,11)
    (agree,10)
    (hereunder,,1)
    (“control”,1)
    (Grant,2)
    (2.2.,2)
    (include,11)
    ... -------------------------------------------
    Time: 1504665719000 ms -------------------------------------------
    Time: 1504665739000 ms -------------------------------------------
    ------------------------------------------- Time: 1504665740000 ms -------------------------------------------
    (under,1)
    (Technology,1)
    (distribution,2)
    (http://hadoop.apache.org/core/,1)
    (Unrestricted,1)
    (740.13),1)
    (check,1)
    (have,1)
    (policies,1)
    (uses,1)
    ... -------------------------------------------
    Time: 1504665741000 ms

--------------------

## 二、自定义Receiver ##

通过继承Receiver，并实现onStart、onStop方法来自定义数据源采集。

**模拟Spark内置的Socket链接数据采集：**

import java.io.{BufferedReader, InputStreamReader}
    import java.net.Socket
    import java.nio.charset.StandardCharsets
    
    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
    
    class CustomReceiver (host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) { 
      override def onStart(): Unit = {
        // Start the thread that receives data over a connection
        new Thread("Socket Receiver") {
          override def run() { receive() }
        }.start()
      }
    
      override def onStop(): Unit = {
        // 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
          socket = new Socket(host, port)
    
          // Until stopped or connection broken continue reading
          val reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.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")
        } 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)
        }
      }
    }
    
    object CustomReceiver { 
      def main(args: Array[String]) {
    
        val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
        val ssc = new StreamingContext(conf, Seconds(1))
    
        // Create a DStream that will connect to hostname:port, like localhost:9999
        val lines = ssc.receiverStream(new CustomReceiver("master01", 9999))
    
        // Split each line into words
        val words = lines.flatMap(_.split(" "))
    
        //import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
        // Count each word in each batch
        val pairs = words.map(word => (word, 1))
        val wordCounts = pairs.reduceByKey(_ + _)
    
        // Print the first ten elements of each RDD generated in this DStream to the console
        wordCounts.print()
    
        ssc.start()             // Start the computation
        ssc.awaitTermination()  // Wait for the computation to terminate
        //ssc.stop()
      }
    }

可以通过streamingContext.receiverStream( < instance of customreceiver >)  
来使用自定义的数据采集源：

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

**和Socket数据流测试方式一致**

--------------------

## 三、RDD队列 ##

**测试过程中，可以通过使用streamingContext.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理。**

import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    import scala.collection.mutable
    
    object QueueRdd { 
    
      def main(args: Array[String]) {
    
        val conf = new SparkConf().setMaster("local[*]").setAppName("QueueRdd")
        val ssc = new StreamingContext(conf, Seconds(1))
    
        // Create the queue through which RDDs can be pushed to
        // a QueueInputDStream
        //创建RDD队列
        val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]()
    
        // Create the QueueInputDStream and use it do some processing
        // 创建QueueInputDStream
        val inputStream = ssc.queueStream(rddQueue)
    
        //处理队列中的RDD数据
        val mappedStream = inputStream.map(x => (x % 10, 1))
        val reducedStream = mappedStream.reduceByKey(_ + _)
    
        //打印结果
        reducedStream.print()
    
        //启动计算
        ssc.start()
    
        // Create and push some RDDs into
        for (i <- 1 to 30) {
          rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
          Thread.sleep(2000)
    
          //通过程序停止StreamingContext的运行
          //ssc.stop()
        }
    
        ssc.awaitTermination()
      }
    }

**执行命令：**

bin/spark-submit --class com.m.jd.streaming.QueueRdd /opt/spark-jar/queueRdd-jar-with-dependencies.jar