在上一篇文章中，我介绍了 Spark Streaming 的基本概念和架构。本文我将进一步探讨 Spark Streaming 的基本输入源，包括文件流、套接字流以及 RDD 队列流。我们可以将这些输入源的数据流转换为 DStream，从而实时处理各种数据。

二、基本输入源 1：文件流（File Streams）

1，基本介绍

2，使用样例

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    // 监控的目录路径
    val directory = "/Volumes/BOOTCAMP/test"

    // 创建文件流，从指定目录读取数据
    val lineStreams = ssc.textFileStream(directory)

    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))

    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))

    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)

    // 打印结果
    wordAndCountStreams.print()

    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止
    ssc.awaitTermination()
  }
}

（2）程序启动后，它会开始监控指定目录，并读取在监控目录中新产生文件的数据。我们可以测试一下，进入设定的监控目录，创建一个文本文件：

cd /Volumes/BOOTCAMP/test
vi hangge.txt

（3）在该文件中添加如下内容，然后保存退出：

hangge.com 123
123 a b c
a

（4）Spark Streaming 应用程序这边将会实时处理该文件的数据并输出结果：

2，使用样例

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    // 通过监控端口创建 DStream，读进来的数据为一行行
    val lineStreams = ssc.socketTextStream("localhost", 9999)

    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))

    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))

    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)

    // 打印结果
    wordAndCountStreams.print()

    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }
}

nc -lk 9999

（3）程序启动后，我们在该终端中输入一些文本数据：

三、基本输入源 3：RDD队列流（RDD Queue Streams）

1，基本介绍

2，使用样例

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val ssc = new StreamingContext(sparkConf, Seconds(3))

    // 创建RDD队列流
    val rddQueue = new mutable.Queue[RDD[String]]()

    // 使用RDD队列流创建DStream
    val inputStream = ssc.queueStream(rddQueue)

    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputStream.flatMap(_.split(" "))

    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))

    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)

    // 打印结果
    wordAndCountStreams.print()

    // 启动 StreamingContext
    ssc.start()

    // 定时向RDD队列中放入新的RDD
    for (i <- 1 to 5) {
      val randomText = generateRandomText()
      val rdd = ssc.sparkContext.parallelize(Seq(randomText))
      rddQueue += rdd
      Thread.sleep(5000) // 每5秒放入一个新的RDD
    }

    // 等待应用程序终止
    ssc.awaitTermination()
  }

  // 生成随机文本
  def generateRandomText(): String = {
    val words = List("hello", "world", "spark", "streaming", "hangge.com")
    val random = new Random()
    val randomWords = (1 to 10).map(_ => words(random.nextInt(words.length)))
    randomWords.mkString(" ")
  }
}

（2）程序启动后运行结果如下：