Structured Streaming简介

文章目录

Structured Streaming 简介
- Spark Streaming vs. Structured Streaming
- 计算模型
- - Batch mode
  - Continuous mode
- 容错机制
- - Batch mode 容错
  - Continuous mode 容错
- Watermark 机制

Structured Streaming 简介

Spark Streaming vs. Structured Streaming

Spark Streaming

Spark Streaming是spark最初的流处理框架，使用了微批的形式来进行流处理。

提供了基于RDDs的Dstream API，每个时间间隔内的数据为一个RDD，源源不断对RDD进行处理来实现流计算

Structured Streaming

Spark 2.X出来的流框架，采用了无界表的概念，流数据相当于往一个表上不断追加行。

基于Spark SQL引擎实现，可以使用大多数Spark SQL的function

计算模型

Trigger 机制：决定引擎在什么时候、以怎样的方式和频率去处理接收到的数据流。Structured Streaming 支持 4 种 Trigger

Structured Streaming 支持两种计算模型，分别是 Batch mode 和 Continuous mode。计算模型本质上，它要解决的问题是 Spark 以怎样的方式，来对待并处理流数据。

Batch mode

Batch mode，指的是 Spark 将连续的数据流切割为微批（Micro-batch）数据，也即小份的数据集。

每一份 Micro-batch，都会触发一个 Spark Job，每一个 Job 会包含若干个Tasks。这些 Tasks 最终会交由 Spark SQL 与 Spark Core 去做优化与执行。

Continuous mode

Continuous mode 并不切割数据流，而是以事件 / 消息（Event / Message）为粒度，用连续的方式来处理数据。这里的事件或是消息，指代的是原始数据流中最细粒度的数据形式，它可以是一个单词、一行文本，或是一个画面帧。

在 Continuous mode 下，Structured Streaming 使用一个常驻作业（Long running job）来处理数据流（或者说服务）中的每一条消息。

**Batch mode 吞吐量大、延迟高（秒级），而 Continuous mode 吞吐量低、延迟更低（毫秒级）。**吞吐量指的是单位时间引擎处理的消息数量，批量数据能够更好地利用 Spark 分布式计算引擎的优势，因此 Batch mode 在吞吐量自然更胜一筹。

容错机制

容错指的是，在计算过程中出现错误（作业层面、或是任务层面，等等）的时候，流处理引擎有能力恢复被中断的计算过程，同时保证数据上的不重不漏，也即保证数据处理的一致性。

从数据一致性的角度出发，这种容错的能力，可以划分为 3 种水平：

At most once：最多交付一次，数据存在丢失的风险；
At least once：最少交付一次，数据存在重复的可能；
Exactly once：交付且仅交付一次，数据不重不漏。

这里的交付，指的是数据从 Source 到 Sink 的整个过程。

Structured Streaming 的容错能力是：“结合幂等的 Sink，Structured Streaming 能够提供 Exactly once 的容错能力”。

在数据处理上，结合容错机制，Structured Streaming 能够提供“At least once”的处理能力。
结合幂等的 Sink，Structured Streaming 可以实现端到端的“Exactly once”容错水平。

Batch mode 容错

在 Batch mode 下，Structured Streaming 利用 Checkpoint 机制来实现容错。在实际处理数据流中的 Micro-batch 之前，Checkpoint 机制会把该 Micro-batch 的元信息全部存储到开发者指定的文件系统路径，比如 HDFS 。当出现作业或是任务失败时，引擎只需要读取这些事先记录好的元信息，就可以恢复数据流的“断点续传”。

每一个 Micro-batch 在被 Structured Streaming 引擎实际处理之前， Checkpoint 机制会先把它的元信息记录到Write Ahead Log（WAL 日志）。Batch mode先写日志，再处理数据

每个 Micro-batch 都会触发一个 Spark 作业，作业与任务的频繁调度会引入计算开销，在运行模式与容错机制的双重延迟下，导致Batch mode延迟较高。

Continuous mode 容错

因为 Continuous mode 没有微批，不会涉及到微批中的延迟，到达 Source 中的消息可以立即被 Structured Streaming 引擎消费并处理。但这同时也带来一个问题，那就是引擎如何把当前的处理进度做持久化，从而为失败重试提供可能。

在 Continuous mode 下，Structured Streaming 利用 Epoch Marker 机制，来实现容错。

对于 Source 中的数据流，Structured Streaming 每隔一定时间（可设置），安插一个 Epoch Marker，而两个 Epoch Marker 之间的数据称为一个 Epoch。

在引擎处理并交付数据的过程中，每当遇到 Epoch Marker 的时候，引擎都会把对应 Epoch 中最后一条消息的 Offset 写入日志，从而实现容错。日志的写入是异步的，因此这个过程不会对数据的处理造成延迟。

Continuous mode先处理数据，然后再写日志

Watermark 机制

Watermark 机制是用来决定，哪些 Late data 可以参与过往窗口状态的更新，而哪些 Late data 则惨遭抛弃。

水印与水位线，对标的都是消息的事件时间。

水印相当于系统当前接收到的所有消息中最大的事件时间。

水位线指的是水印对应的事件时间，减去用户设置的容忍值（记作 T ）。

水位线对应的事件时间，称作 Watermark。

Watermark = max event time - T

当有新消息到达系统后，Structured Streaming 首先判断它的事件时间，是否大于水印。

如果事件时间大于水印的话，Watermark 机制则相应地更新水印与水位线，即最大事件时间与 Watermark。
假设新到消息的事件时间小于当前水印（当前最大事件时间），那么系统进一步判断消息的事件时间与“Watermark 时间窗口下沿”的关系。所谓“Watermark 时间窗口下沿”，它指的是 Watermark 所属时间窗口的起始时间。
- 新到消息的事件时间大于“Watermark 时间窗口下沿”，则消息可以参与过往窗口的状态更新；
- 否则，消息将被系统抛弃，不再参与计算。
Watermark 时间窗口下沿说明

假设 Watermark 为“2021-10-01 09:34:00”，且事件时间窗口大小为 5 分钟，那么，Watermark 所在时间窗口就是[“2021-10-01 09:30:00”，“2021-10- 01 09:35:00”]，也即窗口 30-35。这个时候，“Watermark 时间窗口下沿”，就是窗口 30-35 的起始时间，也就是“2021-10-01 09:30:00”，如下图所示。