大数据 -- Structured Streaming

1. 在Structured Streaming发布之后，DataFrame即可以代表静态的 有边界数据 ，也可以代表 无边界数据
2. 之前对 静态DataFrame 的各种操作同样适用于 流式DataFrame

创建DataFrame

1. SparkSession.readStream返回的DataStreamReader可以用于创建 流DataFrame
2. 支持多种类型的数据流作为输入，例如 file、Kafka、socket 等

socketDataFrame = spark
   .readStream
   .format("socket"）
   .option("host", "localhost")
   .option("port", 9999)
   .load()

基本的查询操作

流DataFrame和静态DataFrame一样，不仅支持 类似SQL的查询操作 ，还 支持RDD的转换操作

df = // 该DataFrame代表学生的数据流，schema是 {name:string, age:number, height:number, grade:string}
df.select("name").where("age > 10") // 返回年龄大于10岁的学生名字列表
df.groupBy("grade").count() // 返回每个年级学生的人数
df.sort_values(["age"], ascending=False).head(100) // 返回100个年龄最大的学生

可以通过isStreaming函数来判断一个DataFrame是否代表流数据

df.isStreaming()

基于 事件时间 的时间窗口操作

words = // 该DataFrame代表词语的数据流，schema是 {timestamp:Timestamp, word:String}

// 基于词语的生成时间，创建一个窗口长度为1分钟，滑动间隔为10秒钟的window
// 然后，把输入的词语根据window和词语本身聚合起来，并统计每个window内每个词语的数量
// 最后，根据词语的数量进行排序，只返回前10的词语
windowedCounts = words.groupBy(
   window(words.timestamp, "1 minute", "10 seconds"),
   words.word
).count()
.sort(desc("count"))
.limit(10)

输出结果流

Dataset.writeStream返回的DataStreamWriter支持多种写入位置，例如file、Kafka、console和内存等

query = wordCounts
   .writeStream
   .outputMode("complete")
   .format("csv")
   .option("path", "path/to/destination/dir")
   .start()

query.awaitTermination()

对比Spark Streaming

综合来说， Structured Streaming是比Spark Streaming更好的流处理工具

易用性和性能

1. Spark Streaming提供的DStream API与RDD API很类似，都是相对 比较底层的API
2. 编写Spark Streaming程序时，本质上是要去 构造RDD的DAG执行图 ，然后 通过Spark Engine运行
   • 开发者的 任务很重 ，需要想办法去提高程序的处理效率
   • 一个好的数据处理框架，开发者只需要 专注于业务逻辑 ，而不用操心配置、优化等繁琐事项
3. Structured Streaming提供的 DataFrame API 是一个 相对高级的API
   • 大部分开发者都很熟悉 关系型数据库 和 SQL
   • 这样的数据抽象可以让开发者 用一套统一的方案去处理批处理和流处理 ，而无需关心具体的执行细节
   • DataFrame API是在 Spark SQL 的引擎上执行的，Spark SQL有很多优化，所以Structured Streaming的程序 性能很好

实时性

1. Spark Streaming是 准实时 的，能做到的最小延迟在 1秒 左右
2. Structured Streaming采用的也是 微批处理 思想，但能做到更小的时间间隔，最小延迟在 100毫秒 左右
3. 在Spark 2.3中，Structured Streaming引入了 连续处理 模式，可以做到真正的 毫秒级延迟