Spark Structured Streaming 入门

无限增长的表格

Structured Streaming 的关键思想是将持续不断的数据当做一个不断追加的表。

将输入的流数据当做一张 “输入表”。把每一条到达的数据作为输入表的新的一行来追加。

image

在输入表上执行的查询将会生成 “结果表”。每个触发间隔(trigger interval)(如: 1s)，新的行追加到输入表，最终更新结果表。无论何时更新结果表，我们都希望将更改的结果行 output 到外部存储/接收器(external sink)。

image

基于“无限增长的表格”的编程模型的 Structured Streaming 的 word count例子：

val spark = SparkSession.builder().master("...").getOrCreate()  // 创建一个 SparkSession 程序入口

val lines = spark.readStream.textFile("some_dir")  // 将 some_dir 里的内容创建为 Dataset/DataFrame；即 input table
val 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() 

image

• Structured Streaming 也是先纯定义、再触发执行的模式，即
  • 前面大部分代码是 纯定义Dataset/DataFrame 的产生、变换和写出
  • 后面位置再真正 start一个新线程，去触发执行之前的定义
• 在新的执行线程里我们需要持续地去发现新数据，进而持续地查询最新计算结果至写出
  • 这个过程叫做 continous query（持续查询)

写入模式

Complete

整个更新的结果表将被写入外部存储。由存储连接器（storage connector）决定如何处理整个表的写入。

image

Append

只有结果表中自上次触发后附加的新行将被写入外部存储。这仅适用于不期望更改结果表中现有行的查询。

image

Append 的语义将保证，一旦输出了某条 key，未来就不会再输出同一个 key。

所以，在上图 12:10 这个批次直接输出 12:00-12:10|cat|1, 12:05-12:15|cat|1 将是错误的，因为在 12:20 将结果更新为了 12:00-12:10|cat|2，但是 Append 模式下却不会再次输出 12:00-12:10|cat|2，因为前面输出过了同一条 key 12:00-12:10|cat 的结果12:00-12:10|cat|1。

为了解决这个问题，在 Append 模式下，Structured Streaming 需要知道，某一条 key 的结果什么时候不会再更新了。当确认结果不会再更新的时候，就可以将结果进行输出。

Update

只有自上次触发后结果表中更新的行将被写入外部存储（自 Spark 2.1.1 起可用）。 请注意，这与完全模式不同，因为此模式仅输出自上次触发以来更改的行。如果查询不包含聚合操作，它将等同于附加模式。

image

如上图所示，在 Update 模式中，只有本执行批次 State 中被更新了的条目会被输出：

• 在 12:10 这个执行批次，State 中全部 2 条都是新增的（因而也都是被更新了的），所以输出全部 2 条；
• 在 12:20 这个执行批次，State 中 2 条是被更新了的、 4 条都是新增的（因而也都是被更新了的），所以输出全部 6 条；
• 在 12:30 这个执行批次，State 中 4 条是被更新了的，所以输出 4 条。这些需要特别注意的一点是，如 Append 模式一样，本执行批次中由于（通过 watermark 机制）确认 12:00-12:10 这个 window 不会再被更新，因而将其从 State 中去除，但没有因此产生输出。

StreamExecution：持续查询的驱动引擎

初始状态

前文刚解析的先定义好 Dataset/DataFrame 的产生、变换和写出，再启动 StreamExection 去持续查询。这些 Dataset/DataFrame 的产生、变换和写出的信息就对应保存在 StreamExecution 非常重要的 3 个成员变量中：

• sources: streaming data 的产生端（比如 kafka 等）
• logicalPlan: DataFrame/Dataset 的一系列变换（即计算逻辑）
• sink: 最终结果写出的接收端（比如 file system 等）

StreamExection 另外的重要成员变量是：

• currentBatchId: 当前执行的 id
• batchCommitLog: 已经成功处理过的批次有哪些
• offsetLog, availableOffsets, committedOffsets: 当前执行需要处理的 source data 的 meta 信息
• offsetSeqMetadata: 当前执行的 watermark 信息（event time 相关）等

image

持续查询

image

一次执行的过程如上图；这里有 6 个关键步骤：

1. StreamExecution 通过 Source.getOffset() 获取最新的 offsets，即最新的数据进度；
2. StreamExecution 将 offsets 等写入到 offsetLog 里
   • 这里的 offsetLog 是一个持久化的 WAL (Write-Ahead-Log)，是将来可用作故障恢复用
3. StreamExecution 构造本次执行的 LogicalPlan
   • (3a) 将预先定义好的逻辑（即 StreamExecution 里的 logicalPlan 成员变量）制作一个副本出来
   • (3b) 给定刚刚取到的 offsets，通过 Source.getBatch(offsets) 获取本执行新收到的数据的 Dataset/DataFrame 表示，并替换到 (3a) 中的副本里
   • 经过 (3a), (3b) 两步，构造完成的 LogicalPlan 就是针对本执行新收到的数据的 Dataset/DataFrame 变换（即整个处理逻辑）了
4. 触发对本次执行的 LogicalPlan 的优化，得到 IncrementalExecution
   • 逻辑计划的优化：通过 Catalyst 优化器完成
   • 物理计划的生成与选择：结果是可以直接用于执行的 RDD DAG
   • 逻辑计划、优化的逻辑计划、物理计划、及最后结果 RDD DAG，合并起来就是 IncrementalExecution
5. 将表示计算结果的 Dataset/DataFrame (包含 IncrementalExecution) 交给 Sink，即调用 Sink.add(ds/df)
6. 计算完成后的 commit
   • (6a) 通过 Source.commit() 告知 Source 数据已经完整处理结束；Source 可按需完成数据的 garbage-collection
   • (6b) 将本次执行的批次 id 写入到 batchCommitLog 里

持续查询（增量）

image

Structured Streaming 在编程模型上暴露给用户的是，每次持续查询看做面对全量数据（而不仅仅是本次执行信收到的数据），所以每次执行的结果是针对全量数据进行计算的结果。

但是在实际执行过程中，由于全量数据会越攒越多，那么每次对全量数据进行计算的代价和消耗会越来越大。

Structured Streaming 的做法是：

• 引入全局范围、高可用的 StateStore
• 转全量为增量，即在每次执行时：
  • 先从 StateStore 里 restore 出上次执行后的状态
  • 然后加入本执行的新数据，再进行计算
  • 如果有状态改变，将把改变的状态重新 save 到 StateStore 里
• 为了在 Dataset/DataFrame 框架里完成对 StateStore 的 restore 和 save 操作，引入两个新的物理计划节点 —— StateStoreRestoreExec 和 StateStoreSaveExec

所以 Structured Streaming 在编程模型上暴露给用户的是，每次持续查询看做面对全量数据，但在具体实现上转换为增量的持续查询。

故障恢复

通过上文知道存储 source offsets 的 offsetLog，和存储计算状态的 StateStore，是全局高可用的。仍然采用前面的示意图，offsetLog 和 StateStore 被特殊标识为紫色，代表高可用。

image

由于 exectutor 节点的故障可由 Spark 框架本身很好的 handle，不引起可用性问题，我们本节的故障恢复只讨论 driver 故障恢复。

如果在某个执行过程中发生 driver 故障，那么重新起来的 StreamExecution：

• 读取 WAL offsetlog 恢复出最新的 offsets 等；相当于取代正常流程里的 (1)(2) 步
• 读取 batchCommitLog 决定是否需要重做最近一个批次
• 如果需要，那么重做 (3a), (3b), (4), (5), (6a), (6b) 步
  • 这里第 (5) 步需要分两种情况讨论
    • (i) 如果上次执行在 (5) *结束前即失效*，那么本次执行里 sink 应该完整写出计算结果
    • (ii) 如果上次执行在 (5) *结束后才失效*，那么本次执行里 sink 可以重新写出计算结果（覆盖上次结果），也可以跳过写出计算结果（因为上次执行已经完整写出过计算结果了）

这样即可保证每次执行的计算结果，在 sink 这个层面，是 *不重不丢* 的 —— 即使中间发生过 1 次或以上的失效和恢复。

参考

Structured Streaming 实现思路与实现概述
Structured Streaming Programming Guide