Spark综合学习笔记（三十二）Structured Streaming Source-Socket

https://www.bilibili.com/video/BV1Xz4y1m7cv?p=70

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html

package structured

import org.apache.spark.sql.{Dataframe, Dataset, Row, SparkSession}


object Demo01_Source_Socket {
  def main(args: Array[String]): Unit = {
    //TODO 0.创建环境
    //因为StructuredStreaming基于SparkSQL的且编程API/数据抽象是Dataframe/DataSet,所以这里创建SparkSession即可
    val spark = SparkSession.builder().appName("sparksql").master("local[*]")
      .config("spark.sql.shuffle.partitions","4").getOrCreate()//本次测试时分区数设置小一点，实际开发中可以根据集群规模设置大小
    val sc = spark.sparkContext
    sc.setLogLevel("WARN")
    import spark.implicits._
    import org.apache.spark.sql.functions._
    //todo.1.加载数据
    val df: Dataframe = spark.readStream.format("socket")
      .option("host", "node1")
      .option("port", 9999)
      .load()
    df.printSchema()
    //df.show()// Queries with streaming sources must be executed with writeStream.start();


    //todo.2.处理数据
    val ds: Dataset[String] = df.as[String]
    val result: Dataset[Row] = ds.flatMap(_.split(" "))
      .groupBy('value)
      .count()
      .orderBy('count.desc)

    //todo.3.输出结果
    result.writeStream
        .format("console")
      .outputMode("complete")

    //todo.4.启动并等待结束
      .start()
      .awaitTermination()
    //todo.5.关闭资源
    spark.stop()
  }

}

（1）在虚拟机node1，监听9999端口
输入一串字符，短暂的延迟后，在控制端看到了单词统计的结果

（2）继续输入第二行字符串，很快控制台自动更新最新的统计结果

（3）输入第三行

通过简单的演示，可以看到，流式处理的实现过程和批处理相差不大，这也就是说，在实际使用，工作当中，比如说公司当前使用批处理想要转换成流式处理，那么头部配置的代码不需要变动，只需要修改一部分，即可实现批处理到流处理的转换