前言
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
