一、Structured Streaming 实现思路与实现概述

矫情吗；* 2024-04-18 15:06 61阅读 0赞

**目录**

一、引言：Spark 2.0 时代

二、从 Structured Data 到 Structured Streaming

三、Structured Streaming：无限增长的表格

--------------------

## **一、引言：Spark 2.0 时代** ##

![format_png][]

Spark 1.x 时代里，以 SparkContext（及 RDD API）为基础，在 structured data 场景衍生出了 SQLContext, HiveContext，在 streaming 场景衍生出了 StreamingContext，很是琳琅满目。

![format_png 1][]

Spark 2.x 则咔咔咔精简到只保留一个 SparkSession 作为主程序入口，以 Dataset/DataFrame 为主要的用户 API，同时满足 structured data, streaming data, machine learning, graph 等应用场景，大大减少使用者需要学习的内容，爽爽地又重新实现了一把当年的 "one stack to rule them all" 的理想。

![format_png 2][]

我们这里简单回顾下 Spark 2.x 的 Dataset/DataFrame 与 Spark 1.x 的 RDD 的不同：

*  Spark 1.x 的 RDD 更多意义上是一个一维、只有行概念的数据集，比如 `RDD[Person]`，那么一行就是一个 `Person`，存在内存里也是把 `Person` 作为一个整体（序列化前的 java object，或序列化后的 bytes）。
 *  Spark 2.x 里，一个 `Person` 的 Dataset 或 DataFrame，是二维行+列的数据集，比如一行一个 `Person`，有 `name:String`, `age:Int`, `height:Double` 三列；在内存里的物理结构，也会显式区分列边界。
    
     *  Dataset/DataFrame 在 API 使用上有区别：Dataset 相比 DataFrame 而言是 type-safe 的，能够在编译时对 AnalysisExecption 报错（如下图示例）:

![format_png 3][]

*   *  Dataset/DataFrame 存储方式无区别：两者在内存中的存储方式是完全一样的、是按照二维行列（UnsafeRow）来存的，所以在没必要区分 `Dataset` 或 `DataFrame` 在 API 层面的差别时，我们统一写作 `Dataset/DataFrame`

> \[注\] 其实 Spark 1.x 就有了 Dataset/DataFrame 的概念，但还仅是 SparkSQL 模块的主要 API ；到了 2.0 时则 Dataset/DataFrame 不局限在 SparkSQL、而成为 Spark 全局的主要 API。

## **二、从 Structured Data 到 Structured Streaming** ##

使用 Dataset/DataFrame 的行列数据表格来表达 structured data，既容易理解，又具有广泛的适用性：

*  Java 类 `class Person { String name; int age; double height}` 的多个对象可以方便地转化为 `Dataset/DataFrame`
 *  多条 json 对象比如 `{name: "Alice", age: 20, height: 1.68}, {name: "Bob", age: 25, height: 1.76}` 可以方便地转化为 `Dataset/DataFrame`
 *  或者 MySQL 表、行式存储文件、列式存储文件等等等都可以方便地转化为 `Dataset/DataFrame`

Spark 2.0 更进一步，使用 Dataset/Dataframe 的行列数据表格来扩展表达 streaming data —— 所以便横空出世了 Structured Streaming 、《Structured Streaming 源码解析系列》—— 与静态的 structured data 不同，动态的 streaming data 的行列数据表格是一直无限增长的（因为 streaming data 在源源不断地产生）！

![format_png 4][]

## 三、Structured Streaming：无限增长的表格 ##

基于“无限增长的表格”的编程模型 \[1\]，我们来写一个 streaming 的 word count：

![format_png 5][]

对应的 Structured Streaming 代码片段：

val spark = SparkSession.builder().master("...").getOrCreate()  // 创建一个 SparkSession 程序入口
    val lines = spark.readStream.textFile("some_dir")  // 将 some_dir 里的内容创建为 Dataset/DataFrame；即 input tableval words = lines.flatMap(_.split(" "))val wordCounts = words.groupBy("value").count()    // 对 "value" 列做 count，得到多行二列的 Dataset/DataFrame；即 result table
    val query = wordCounts.writeStream                 // 打算写出 wordCounts 这个 Dataset/DataFrame  .outputMode("complete")                          // 打算写出 wordCounts 的全量数据  .format("console")                               // 打算写出到控制台  .start()                                         // 新起一个线程开始真正不停写出
    query.awaitTermination()                           // 当前用户主线程挂住，等待新起来的写出线程结束

这里需要说明几点：

*  Structured Streaming 也是先纯定义、再触发执行的模式，即
    
     *  前面大部分代码是 ***纯定义*** Dataset/DataFrame 的产生、变换和写出
     *  后面位置再真正 ***start*** 一个新线程，去触发执行之前的定义

*  在新的执行线程里我们需要 ***持续地*** 去发现新数据，进而 ***持续地*** 查询最新计算结果至写出
    
     *  这个过程叫做 ***continous query（持续查询）***

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzYkwzc2ljaWJieWZBb3hJUjJJaldyenlGSFZpYkg5clNpYW1ZSmlidmNvWHRkWXQyV25FSHF4aWJ4V2hBLzY0MD90cD13ZWJwJnd4ZnJvbT01Jnd4X2xhenk9MSZ3eF9jbz0x?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzY2ljU2FWTnFKUHlua2RlZTBGM3pwNElGNmg2VDZmRFVxZkVCdlNyTkV0T0VUY0xHWnlBd3Rody82NDA_dHA9d2VicCZ3eGZyb209NSZ3eF9sYXp5PTEmd3hfY289MQ?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzclhSQkdYOGRFQkFzUjdMdzZlWnhkV21kZm9oVElDNGMyb000ZE1pYmN0SzdJbmliOFU4UEFCbWcvNjQwP3RwPXdlYnAmd3hmcm9tPTUmd3hfbGF6eT0xJnd4X2NvPTE?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzR01YSWRWa2FpYTZyUllSVEdiOWNOWUY4MVVGanI2NWdPNU5VNDNkNWdkVmliczJsQ21scVdCWncvNjQwP3RwPXdlYnAmd3hmcm9tPTUmd3hfbGF6eT0xJnd4X2NvPTE?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzeWZZbXhlN0xpYjJXcVRNNEk0Z2hFN1d1VXZlZ3RoQWZEQWlid1ZBbWRNSjkwTXlNT0xjQ1B0T1EvNjQwP3RwPXdlYnAmd3hmcm9tPTUmd3hfbGF6eT0xJnd4X2NvPTE?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9VUzEwR2NkMHRRSFZFSGlheGljd0NsblRyTHFveUJOTUgzdVdPOWljeVpQcXRUNlYzMnpYalhra0J4eExhcDY5UGZCd1JrWjhrV1BtMGgxY0ZMNTNNYzVwdy82NDA_dHA9d2VicCZ3eGZyb209NSZ3eF9sYXp5PTEmd3hfY289MQ?x-oss-process=image/format,png