前言

Structured Streaming的数据源目前支持File Source , Kafka Source

Socket Source ,Rate Source主要是测试用

一、Input Source

监视目录的文件改变，只能是原子性的改变，比如把文件放入该目录，而不是持续写入该目录中的某个文件。

File Source

demo1源码-txt文件

/**
  *  Structured Streaming监控目录 text格式数据
  */
object SSReadTextData {
  def main(args: Array[String]): Unit = {

    //1.创建对象
    val spark: SparkSession = SparkSession.builder().master("local")
      .appName("SSReadTextData")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import  spark.implicits._

    spark.sparkContext.setLogLevel("Error")

    //2.监控目录
    val ds: Dataset[String] = spark.readStream.textFile("./datatxt/")

    val result: DataFrame = ds.map(line => {
      val arr: Array[String] = line.split("-")
      (arr(0).toInt, arr(1), arr(2).toInt)
    }).toDF("id", "name", "age")

    val query: StreamingQuery = result.writeStream
      .format("console")
      .start()

    query.awaitTermination()

  }

}

运行demo后，把 test1.txt,test2.txt 拖入datatxt目录，能观察到idea console两次打出批次计算数据

[图片上传失败...(image-dcd0f-1678378250033)]

-------------------------------------------
Batch: 0
-------------------------------------------
+---+-----+---+
| id| name|age|
+---+-----+---+
|  1|name1| 18|
|  2|name2| 19|
|  3|name3| 20|
+---+-----+---+

-------------------------------------------
Batch: 1
-------------------------------------------
+---+-----+---+
| id| name|age|
+---+-----+---+
|  1|name1| 21|
|  2|name2| 22|
|  3|name3| 23|
+---+-----+---+

demo2源码-csv文件

传入 structType参数。
指定分隔符
readStream.csv

//2.创建CSV数据schema
val userSchema: StructType = new StructType().add("id", "integer")
  .add("name", "string")
  .add("age", "integer")

val result: DataFrame = spark.readStream
  .option("sep", ",")
  .schema(userSchema)
  .csv("./datacsv/")

demo2源码-json文件

传入 structType参数schema和jsonkey要一致 , 不需要指定分隔符。
readStream.json

  //2.创建 json 数据schema
    val userSchema: StructType = new StructType().add("id", "integer")
      .add("name", "string")
      .add("age", "integer")
      .add("gender", "string")


    val result: DataFrame = spark.readStream
      .schema(userSchema)
      .json("./datajson/")

Rate Source

每个输出行包含一个timestamp和value，其中timestamp是一个Timestamp含有信息分配的时间类型，value是从0开始的Long类型的数据，Rate Source式多用于测试

/**
  * SSRateSource
  */
object SSRateSource {
  def main(args: Array[String]): Unit = {
    //1.创建对象
    val spark: SparkSession = SparkSession.builder().master("local")
      .appName("rate test")
//      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val result: DataFrame = spark.readStream
      .format("rate")
      // 配置每秒生成多少行数据，默认1行
      .option("rowsPerSecond", "10")
      .option("numPartitions", 5)
      .load()
    result.writeStream
      .format("console")
      .option("numRows","100")
      .option("truncate","false")
      .start()
      .awaitTermination()

  }
}

输出：

-------------------------------------------
Batch: 22
-------------------------------------------
+-----------------------+-----+
|timestamp              |value|
+-----------------------+-----+
|2023-03-09 15:30:27.144|220  |
|2023-03-09 15:30:27.644|225  |
|2023-03-09 15:30:27.244|221  |

二、Output Sink

数据可以写出到File，Kafka，console，内存。

checkpoint

对于一些可以保证端到端容错的仅仅消费一次的语义场景，需要指定checkpoint目录来写入数据信息，指定的checkpoint目录可以是HDFS中的某个路径（测试时用本地文件系统）.

checkpoint目录中会有以下目录及数据：

offsets：记录偏移量目录，记录了每个批次的偏移量。
commits:记录已经完成的批次，方便重启任务检查完成的批次与offset批次做对比，继续offset消费数据，运行批次。
metadata：metadata元数据保存jobid信息。
sources:数据源各个批次读取详情。
sinks:数据sink写出批次情况。
state:记录状态值，例如：聚合、去重等场景会记录相应状态，会周期性的生成snapshot文件记录状态。

file sink

必须设置checkpoint

/**
  *  读取Socket数据，将数据写入到csv文件
  */
object FileSink {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local")
      .appName("File Sink")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val result: DataFrame = spark.readStream
      .format("socket")
      .option("host", "node1")
      .option("port", 9999)
      .load()

    val query: StreamingQuery = result.writeStream
      .format("csv")
      .option("path", "./dataresultlyw/csvdir")
      .option("checkpointLocation","./checkpintlyw/dir3")
      .start()
    query.awaitTermination()

  }
}

structured_stream_output_1.png

structured_stream_output_2.png

memory sink

测试用，写入到内存表中，用sql获取。

三、Foreach Sink

foreachBatch

一批次数据进行处理，自定义保存到如mysql。

foreach

一条一条处理数据

structured_stream_intput_1.png

大数据学习之(二十)spark structured streaming inputsource