Structured Streaming
编程模型
Structured Streaming 的关键思想是将持续不断的数据当做一个不断追加的表。
基本概念
将输入的流数据当做一张 “输入表”。把每一条到达的数据作为输入表的新的一行来追加。
在输入表上执行的查询将会生成 “结果表”。每个触发间隔（trigger interval）（例如 1s），新的行追加到输入表，最终更新结果表。无论何时更新结果表，我们都希望将更改的结果行 output 到外部存储/接收器（external sink）。
output 有以下三种模式：
- Complete Mode：整个更新的结果表将被写入外部存储。由存储连接器（storage connector）决定如何处理整个表的写入
- Append Mode：只有结果表中自上次触发后附加的新行将被写入外部存储。这仅适用于不期望更改结果表中现有行的查询。
- Update Mode：只有自上次触发后结果表中更新的行将被写入外部存储（自 Spark 2.1.1 起可用）。 请注意，这与完全模式不同，因为此模式仅输出自上次触发以来更改的行。如果查询不包含聚合操作，它将等同于附加模式。
请注意，每种模式适用于某些类型的查询。
为了说明这个模型的使用，让我们来进一步理解上面的快速示例：
- 最开始的 DataFrame lines 为输入表
-
最后的 DataFrame wordCounts 为结果表
在流上执行的查询将 DataFrame lines 转化为 DataFrame wordCounts 与在静态 DataFrame 上执行的操作完全相同。当启动计算后，Spark 会不断从 socket 连接接收数据。如果有新的数据到达，Spark将运行一个 “增量” 查询，将以前的 counts 与新数据相结合，以计算更新的 counts，如下所示：
这种模式与许多其他流处理引擎有显著差异。许多流处理引擎要求用户自己维护运行的状态，因此必须对容错和数据一致性（at-least-once, or at-most-once, or exactly-once）进行处理。 在这个模型中，当有新数据时，Spark负责更新结果表，从而减轻用户的工作。
流式 DataFrames/Datasets 上的操作
可以在流式 DataFrames/Datasets 上应用各种操作：从无类型，类似 SQL 的操作（比如 select、where、groupBy），到类似有类型的 RDD 操作（比如 map、filter、flatMap）
基本操作 - Selection, Projection, Aggregation
大部分常见的 DataFrame/Dataset 操作也支持流式的 DataFrame/Dataset。
case class DeviceData(device: String, deviceType: String, signal: Double, time: DateTime)
val df: DataFrame = ... // streaming DataFrame with IOT device data with schema { device: string, deviceType: string, signal: double, time: string }
val ds: Dataset[DeviceData] = df.as[DeviceData] // streaming Dataset with IOT device data
// Select the devices which have signal more than 10
df.select("device").where("signal > 10") // using untyped APIs
ds.filter(_.signal > 10).map(_.device) // using typed APIs
// Running count of the number of updates for each device type
df.groupBy("deviceType").count() // using untyped API
// Running average signal for each device type
import org.apache.spark.sql.expressions.scalalang.typed
ds.groupByKey(_.deviceType).agg(typed.avg(_.signal)) // using typed API
event-time（事件时间）上的 window 操作
使用 Structured Streaming 进行滑动的 event-time 窗口聚合是很简单的，与分组聚合非常类似。在分组聚合中，为用户指定的分组列中的每个唯一值维护一个聚合值（例如计数）。在基于 window 的聚合的情况下，为每个 window 维护聚合（aggregate values），流式追加的行根据 event-time 落入相应的聚合。让我们通过下图来理解。
想象下，我们的快速示例现在改成了包含数据生成的时间。现在我们想在 10 分钟的 window 内计算 word count，每 5 分钟更新一次。比如 12:00 - 12:10, 12:05 - 12:15, 12:10 - 12:20 等。12:00 - 12:10 是指数据在 12:00 之后 12:10 之前到达。现在，考虑一个 word 在 12:07 的时候接收到。该 word 应当增加 12:00 - 12:10 和 12:05 - 12:15 相应的 counts。所以 counts 会被分组的 key 和 window 分组。
结果表将如下所示：
由于这里的 window 与 group 非常类似，在代码上，你可以使用 groupBy 和 window 来表达 window 聚合。例子如下：
import spark.implicits._
val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
// Group the data by window and word and compute the count of each group
val windowedCounts = words.groupBy(
window($"timestamp", "10 minutes", "5 minutes"),
$"word"
).count()
Watermark 和延迟数据处理
现在考虑一个数据延迟到达会怎么样。例如，一个在 12:04 生成的 word 在 12:11 被接收到。application 会使用 12:04 而不是 12:11 去更新 12:00 - 12:10的 counts。这在基于 window 的分组中很常见。Structured Streaming 会长时间维持部分聚合的中间状态，以便于后期数据可以正确更新旧 window 的聚合，如下所示：
然后，当 query 运行了好几天，系统必须限制其累积的内存中中间状态的数量。这意味着系统需要知道什么时候可以从内存状态中删除旧的聚合，因为 application 不会再为该聚合更晚的数据进行聚合操作。为启动此功能，在Spark 2.1中，引入了 watermark（水印），使引擎自动跟踪数据中的当前事件时间，并相应地清理旧状态。你可以通过指定事件时间列来定义一个 query 的 watermark 和 late threshold（延迟时间阈值）。对于一个开始于 T 的 window，引擎会保持中间状态并允许后期的数据对该状态进行更新直到 max event time seen by the engine - late threshold > T。换句话说，在延迟时间阈值范围内的延迟数据会被聚合，但超过该阈值的数据会被丢弃。让我们以一个例子来理解这一点。我们可以使用 withWatermark() 定义一个 watermark，如下所示：
import spark.implicits._
val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
// Group the data by window and word and compute the count of each group
val windowedCounts = words
.withWatermark("timestamp", "10 minutes")
.groupBy(
window($"timestamp", "10 minutes", "5 minutes"),
$"word")
.count()
在这个例子中，我们定义了基于 timestamp 列定义了 watermark，并且将 10 分钟定义为允许数据延迟的阈值。如果该数据以 update 输出模式运行：
- 引擎将不断更新结果表中 window 中的 counts 直到该 window 比 watermark 更旧
-
数据中的 timestamp 值比当前的最大 event-time 落后 10 分钟以上的数据将被丢弃
以下为示图：
如图所示，引擎跟踪的最大 event-time 是蓝色虚线，并且在每个 trigger 开始时设置 watermark 为 (max event time - '10 mins') 的红线例如，当引擎发现 (12:14, dog) 时将下次 trigger 的 watermark 设置为 12:04。然后，当 watermark 更新为 12:11 时，window （12:00 - 12:10） 的中间状态被清除，所有后续数据（例如（12:04，donkey））被认为是“太晚”，因此被丢弃。根据 output 模式，每次触发后，更新的计数（即紫色行）都将作为触发输出进行写入到 sink。
某些 sink（例如文件）可能不支持 update mode 所需的细粒度更新。所以，我们还支持 append 模式，只有最后确定的计数被写入。这如下图所示。
注意，在非流式 Dataset 上使用 withWatermark 是无效的空操作。
与之前的 update mode 类似，引擎维护每个 window 的中间计数。只有当 window < watermark 时才会删除 window 的中间状态数据，并将该 window 最终的 counts 追加到结果表或 sink 中。例如，window 12:00 - 12:10 的最终结果将在 watermark 更新到 12:11 后再追加到结果表中。
watermark 清除聚合状态的条件十分重要，为了清理聚合状态，必须满足以下条件（自 Spark 2.1.1 起，将来可能会有变化）：
- output mode 必须为 append 或 update：complete mode 需要保留所有的聚合数据，因此 watermark 不能用来清理聚合数据
- 聚合必须具有 event-time 列或基于 event-time 的 window
- withWatermark 必须调用在用来聚合的时间列上。比如 df.withWatermark("time", "1 min").groupBy("time2").count() 是无效的
- withWatermark 必须在调用聚合前调用来说明 watermark 的细节。比如，df.groupBy("time").count().withWatermark("time", "1 min") 是无效的
Output Modes
有几种类型的输出模式：
Append mode（默认的）：这是默认模式，其中只有从上次触发后添加到结果表的新行将被输出到 sink。适用于那些添加到结果表中的行从不会更改的查询。只有 select、where、map、flatMap、filter、join 等查询会支持 Append mode
Complete mode：每次 trigger 后，整个结果表将被输出到 sink。聚合查询（aggregation queries）支持该模式
Update mode：（自 Spark 2.1.1 可用）。只有结果表中自上次 trigger 后更新的行将被输出到 sink
不同类型的流式 query 支持不同的 output mode。以下是兼容性：
