二、Rate 输入源

1，基本介绍

2，使用样例

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("Hello")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 创建一个流式DataFrame，这里从rate读取数据
    val rows = spark.readStream
      .format("rate") // 设置数据源为 rate
      .option("rowsPerSecond", 3) // 设置每秒产生的数据的条数, 默认是 1
      .option("rampUpTime", 1) // 设置多少秒到达指定速率 默认为 0
      .option("numPartitions", 2) /// 设置分区数  默认是 spark 的默认并行度
      .load

    // 启动查询, 把结果打印到控制台
    val query = rows.writeStream
      .outputMode("append") // 使用complete输出模式
      .format("console")
      .start()

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

    //关闭 Spark
    spark.stop()
  }
}

（2）程序启动后，控制台输出内容如下：

三、Socket 输入源

1，基本介绍

2，样例代码

（1）下面代码我们通过 Structured Streaming 读取 9999 端口数据并统计不同单词出现的次数：

import org.apache.spark.sql.SparkSession

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("HelloStructuredStreaming")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 创建一个流式DataFrame，这里从Socket读取数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()

    // 对流式DataFrame进行转换，将每一行数据切分成单词
    val words = lines.as[String].flatMap(_.split(" "))

    // 对单词进行分组和聚合
    val wordCounts = words.groupBy("value").count()

    // 启动查询, 把结果打印到控制台
    val query = wordCounts.writeStream
      .outputMode("complete") // 或者选择其他适合的输出模式
      .format("console")
      .start()

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

    //关闭 Spark
    spark.stop()
  }
}

3，运行测试

nc -lk 9999

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