Streaming Systems: Concept

系统管理员 2023-01-21 12:27 2阅读 0赞

扫码关注公众号免费阅读全文：**冰山烈焰的黑板报**  
![在这里插入图片描述][20200906232656299.jpg_pic_center]

流式数据处理在当今大数据领域是非常重要，这是有足够充分的理由的，如下：

1.  企业需要更及时地洞察他们的数据，而流式数据是实现更低延迟的一个好方法；
2.  现在商业中有海量无界的数据，使用为永不结束的数据设计的系统处理它们就更为容易；
3.  当数据一到达就进行处理，工作负载会随着时间推移更加均匀地分布，从而产生更一致和可预测的资源消耗。

# 1. 术语：Streaming 是什么？ #

在讨论可能遇到的不同类型的数据时，精确的术语也是很有用的。通过两个重要且正交的维度对数据可以唯一确定——Cardinality（基数）和 Constitution（结构）。

数据集的**基数（Cardinality）** 决定其大小，其最主要的方面是数据集是有限的，还是无限的。这有两个描述数据集粗略**基数**的术语：

*  **有界数据（Bounded Data）**：一种有限大小的数据集。例如，HDFS 上的某个文件。
 *  **无界数据（Unbounded Data）**：一种无限大小的数据集，至少理论上是无限的。比如，从手机或传感器发送的信号数据，交易系统中的交易数据等，我们无法预测它什么时候会结束，犹如长江、黄河之水滔滔不绝。

当然，如果我们取交易系统中 2021 年 5 月 16 日的交易数据，那么此时交易数据就变成了有界数据。显而易见，有界数据是无界数据的子集。

另一方面，数据集的**结构（Constitution）**，决定了它的物理表现形式。因此，**结构**定义数据的交互方式。有两种非常重要的结构：

*  **Table**：在特定时间点上的整体视图。Table 一般是用 SQL 进行处理的。
 *  **Stream**：数据集随着时间推移的逐个元素逐个元素（element-by-element）变化的视图。

相比 Batch，Stream 需要做两件事：

1.  **正确性（Correctness）**：这一点 Batch 和 Stream 是相同的。**正确性**可以归结为一致性存储。Streaming System 需要一个随着时间推移 Checkpoint 持久化 State 的方法，并且鉴于机器故障，它必须设计得足够好用以维护一致性。需要强调的是：强一致性是 Exactly-Once 处理所必需的，它是正确性的前提，这也是追上或超越 Batch System 能力的所需的。
2.  **时间推理工具（Tools for reasoning about time）**：它超出了 Batch 的范畴。良好的时间推测工具对于处理变化的时间事件偏差的无界、无序数据是至关重要。

接下来，我们先厘清**时间域（Time Domain）** 的基本概念，之后在进一步分析变化的事件时间上的无界、无序数据是什么。

# 2. Event Time vs Processing Time #

为了确切地讨论无界数据，还需要对时间域有一个清晰的认识。对于任何数据处理系统，我们通常会关心两种时间：

1.  **Event Time（事件时间）**：指事件发生的时间。
2.  **Processing Time（处理时间）**：指系统处理事件的时间。

如果大家对 Flink 有所了解，那么会发现 Flink 还引入了 **Ingestion Time（摄入时间）**，它是指数据到达 Flink 的时间。很少使用这种时间。

当然，不是所有的数据处理，都需要考虑时间因素。但是，大多数场景是需要考虑时间的，比如平台每小时销售额等。理想情况下，Event Time 和 Processing Time 应该是一直相等的。但现实却并非如此，造成二者差异的因素有以下：

1.  共享资源限制，如网络拥塞、或者 CPU 等；
2.  软件因素，如分布式系统逻辑、冲突等；
3.  数据自身的特性，像按 Key 分区，吞吐的变化等。

现实中 Event Time 和 Processing Time 的差异大概如下图：  
![图 1 Event Time 和 Processing Time 的差异][1 Event Time _ Processing Time]  
图中黑色虚线表示理想情况下，Processing Time 和 Event Time 是相等的。橘红色的曲线表示实际上 Processing Time 和 Event Time 的差异。其中系统刚开始时的 Processing Time 有些延迟，数据处理的中间阶段二者趋于一致，行将结束之际又出现延迟。

*  Processing Time：理想虚线和橘红色曲线间的竖直距离，表示事件发生和事件被处理之间的延迟是多少。
 *  Event Time：理想虚线和橘红色曲线间的水平距离，表示数据处理落后 Event Time 多少。

关于延迟/滞后的重要结论是：因为整个 Event Time 和 Processing Time 之间的映射关系不是静态的，所以我们不能仅仅根据 Processing Time 去分析。为了处理无界数据，数据处理系统提供了一个窗口的概念，我们在下文进行阐述。

如果你关心正确性和利用 Event Time 来处理数据，你就不能使用 Processing Time 定义数据的临时分界线。由于这两种时间没有可预测的关系，所以某些 Event Time 的数据可能会划分到错误的 Processing Time 窗口。

不幸的是，即使按照 Event Time 开窗，也不见得一定正确。在无界数据中，无序和不可预知的滞后导致了 Event Time 窗口的数据完整性问题：由于在处理时间和事件时间之间缺乏可预测的映射关系，如何确定何时观察到了给定 Event Time X 的所有数据呢？对于现实中的很多数据源，你都无法确定。

我们与其尝试将无界数据整理成最终变得完整的有限批次，不如让我们接受这些不确定性，设计一套工具。当新数据将到达，旧数据可能会被撤回或更新，我们构建的任何系统都应该能够自己处理这些事实，完整性的概念是对特定和适当用例的一种优化，而不是跨语义的必要性他们都是。

在讨论这种方案实现之前，我们先看看通用数据处理模式。

# 3. Data Processing Patterns #

## 3.1 有界数据 ##

有界数据就是通常意义下的 Batch，处理有界数据概念上很直观，也被我们所熟悉。它是指将一个充满熵的数据集，经过某个数据处理引擎，如 MapReduce，最终产生一种新的结构化数据集。  
![图2 有界数据处理][2]

## 3.2 无界数据：Batch ##

批处理系统虽然设计时并未考虑无界数据，但是它已经被用于处理无界数据了。如你所料，它将无界数据分割成适合批处理的有界数据的集合。

### 3.2.1 Fixed Windows（固定窗口） ###

大多数情况下，将输入的无界数据用固定大小的窗口分割为相互独立的有界数据，然后用批处理系统重复计算。尤其是对于日志这样的数据源，可以自然而然地将其按时间拆成一个树状结构，比如天级日志。

但是，实际上绝大多数系统还要解决数据完整性的问题。比如，如果由于网络分区导致数据延迟达到该怎么办？如果数据是在全球范围内收集的，并且必须在处理前传输到公共位置如何解决？这些都意味着我们必须要采用一些方法来解决，比如，直到延迟的数据都已经收集完之后再处理，或者一旦窗口中有延迟数据到达就重新处理整个批次。  
![图3 无界数据分割到固定大小的窗口][3]

### 3.2.2 Session（会话） ###

固定窗口的批处理方法用于处理会话（Session）这种更复杂的窗口策略时，就会失效。会话是指一个连续的活动周期，由一个不活动的间隔所终止。类似于我们平时访问某些系统时，如果过一段时间不操作，就会重新登录。使用一般的批处理系统计算会话时，会出现一个会话被分到不同的窗口中，如图 4 中红色标记所示。我们可以通过增加批次大小来减少分割次数，但这会增加延迟的成本。另一个方案是添加额外的逻辑，来拼接之前的会话，但这就增加了复杂度。  
![图 4 用固定窗口分割会话][4]  
由此可见，用任何一种传统的批处理系统计算会话都是不理想的。一个更好的方法是，在流上建立会话，在下文我们将会谈到。

## 3.3 无界数据：Streaming ##

与大多数基于批处理的无界数据处理方法相反，流式系统（Streaming Systems）是为无界数据而生的。正如我们之前谈到的，大多数分布式数据源不仅仅是无界数据，而且还具有以下特性：

*  在事件时间（Event Time）方式时高度乱序的，意味着可能会涉及处理乱序的问题。
 *  不同的事件时间偏差，这意味着我们不能假设始终在一个固定的时间范围内，可以看到给定事件时间 X 的大部分数据。

根据无界数据的特点，可以将其处理方式划分为 4 类：

1.  Time-agnostic
2.  Approximation algorithm
3.  Windowing by processing time
4.  Windowing by event time

### 3.3.1 Time-agnostic（时间无关） ###

Time-Agnostic 的处理方式用于根本与时间无关的场景，所有相关的逻辑都是数据驱动。这种处理流式系统除了数据传输，没有什么特别需要支持的。下面让我们看几个例子：

#### Filtering（过滤） ####

过滤是最基本的时间无关处理。当一条数据达到，我们只需考虑它是否是我们感兴趣的，然后过滤掉不感兴趣的数据。因为这种事情在任何时间只与数据自身有关，所以与数据源是无界的、无序的、以及事件时间延迟都无关。  
![图 5 过滤无界数据][5]

#### Inner Join ####

Inner Join 是另一个时间无关的例子。当 Join 两个无界数据源时，我们只关注 Join 的结果，计算逻辑中没有时间元素。当一个数据源的数据达到时，我们可以把它缓存起来，只有当另一个数据源的数据到达时，才去生成 Join 的结果。但是，对于没有发生的 Join 的数据，可能需要采取一些垃圾回收策略，这就和时间有关了。然而，对于很少或者没有未完成 Join 的用例，这种事情就不是问题了。  
![图 6 Inner Join 无界数据][6 Inner Join]  
反观另一个语义 Outer Join，它却涉及数据完整性的问题。当我们观察到 Join 的一方到达后，如何知道另一方是否到达呢？事实上，我们无法确定，这就需要引入一些超时的概念，因此也就引入了时间元素。其实，时间元素本质上是一种窗口形式，后面的文章我们会进一步探讨。

### 3.3.2 Approximation algorithm（近似算法） ###

近似算法也是时间无关的，例如，Top-N，K-means 等。它们消费无界数据，然后生成结果数据。近似算法具有以下优缺点：

*  优点：开销很低，专为无界数据而设计。
 *  缺点：它们的数量有限，算法本身通常很复杂，其近似性质限制了其实用性。  
    ![图 7 近似算法处理无界数据][7]  
    值得注意的是，这些近似算法通常有时间元素，如内置衰减。但因为它们是在数据到达时就处理，所以采用的是处理时间（Processing Time）。近似算法本质上是时间无关的，因此使用起来比较简单。

### 3.3.3 Windowing ###

接下来的两种是关于窗口化的变种。在之前我们已经有过简单的接触，窗口化只是获取数据源（无界或有界），并沿时间边界将其切割成有限块进行处理的概念。  
![图 8 三种窗口][8]  
现在我们先看看三种不同的窗口化策略：

1.  **Fixed/Tumbling Window（固定窗口）**：固定窗口将时间分成具有固定大小时间长度的段。 固定窗口的数据段在整个数据集上是统一的，这是对齐窗口的一个例子。在某些情况下，需要对数据的不同子集（例如，每个键）的窗口进行移动，以随着时间的推移更均匀地分布到窗口完成负载，这是未对齐窗口的一个示例，因为它们随数据变化而变化。
2.  **Sliding/Hopping Window（滑动窗口）**：滑动窗口是固定窗口的一种特例，它具有一个固定的窗口长度和一个固定的滑动周期。如果滑动周期小于窗口长度，窗口彼此之间就有重叠；如果滑动周期等于窗口长度，它就是固定窗口；如果滑动周期大于窗口长度，窗口之间就间隔，只能看到一段时间内的数据子集。与固定窗口一样，滑动窗口通常是对齐的，尽管在某些用例中它们可以不对齐作为性能优化。
3.  **Session Window（会话窗口）**：会话窗口是动态窗口的一个例子，会话由一系列事件组成，这些事件由大于某个超时的不活动间隙终止。 会话通常用于通过将一系列时间相关的事件（例如，一次观看的视频序列）组合在一起来分析用户随时间的行为。 会话窗口很特别，因为它们的长度不能先验定义； 它们取决于所涉及的实际数据。 它们也是未对齐窗口的典型示例，因为会话在不同的数据子集（例如，不同的用户）中实际上永远不会相同。

值得一提的是，在 SQL-based Streaming System 中还会遇到另一种窗口化时间域：**Tuple-Based Windowing**（以包含元素数量计的窗口）。但是，**Tuple-Based Windowing** 本质上是 Processing Time 窗口的另一种形式，其中元素在到达系统时被分配单调递增的时间戳。这就类似于 Flink 中的 Count Window。

我们前面讨论过两种时间域：Processing Time 和 Event Time。而窗口对于这两种时间域都是有意义的。现在我们就来看看它们的异同，首先从基于 Processing Time 的窗口开始吧。

#### Windowing by processing time ####

当用 Processing Time 开窗时，系统实质上会将输入数据缓冲到窗口中，直到经过了一段的 Processing Time 为止。例如，对于 5 分钟的固定窗口，系统将按 Processing Time 缓存 5 分钟的数据，然后将这 5 分钟内接收到的数据视为一个窗口，并将其下发到下游。  
![图 9 基于 Processing Time 的窗口][9 _ Processing Time]

基于 Processing Time 的窗口有以下几个优势：

1.  实现简单。当接收到数据时，缓存数据；当窗口关闭时，发送数据到下游。
2.  判断窗口完成很直接。用处理时间开窗时，不需要处理延迟的数据。
3.  用基于 Processing Time 的窗口可以推测数据源的信息。比如，追踪每秒发送到全局 Web 服务的请求数。

基于 Processing Time 的窗口也有一个最大的劣势是：如果所讨论的数据具有关联的事件时间，这些数据必须以 Event Time 顺序到达，基于 Processing Time 的窗口无法反映这些事件实际发生的时间的事实。

#### Windowing by event time ####

当需要观察数据源中的有限一部分，以反映那些事件实际发生的时间时，可以使用 Event Time 窗口。在 2016 年之前，大多数使用的数据处理系统都缺乏对它的原生支持，虽然任何强一致性系统经过一些修改时能够解决的，比如 Hadoop 和 Spark Streaming 1.x。  
![图 10 基于 Event Time 的固定窗口][10 _ Event Time]  
上图中的黑色箭头指向的两个数据，它们达到的 Processing Time 窗口和它们所属的 Event Time 窗口是不一样的。因此，如果在某个关注 Event Time 的场景下，却使用了 Processing Time 窗口来计算，那么得到的结果就会是错误的。如我们所料，使用 Event Time 窗口可以保证数据事件时间的正确性。

关于无界数据源上的事件时间窗口的另一个好处是，可以创建动态大小的窗口，例如会话，而不会出现在固定窗口上生成会话时产生的任意拆分。  
![图 11  基于 Event Time 的会话窗口][11 _ Event Time]  
任何事情都犹如硬币的两面，Event Time 窗口的语义固然强大，但是它也有两个明显的缺陷：

1.  **Buffering**。由于更长的窗口生命周期，需要更多缓存更多的数据。
2.  **Completeness**。我们不知道是否处理完了窗口中的全部数据，以及如何知道窗口的结果何时可以物化。对于很多类型的输入源，系统可以通过类似 Watermark 的方式，对 Window 的完成给出合理且准确的启发式评估。

# 4. 总结 #

本文介绍了很多内容：澄清术语，介绍完整性和时间工具两个重要概念，阐明 Processing Time 和 Event Time 的关系，分析四种常用的数据处理方法。

[20200906232656299.jpg_pic_center]: /images/20221020/61c576e6a03042459d5f5c6da0ea6d7e.png
[1 Event Time _ Processing Time]: /images/20221020/5c4340b509614ac8bd39d0656e75bbc7.png
[2]: /images/20221020/d523f9ab271447d69b9adc18115260b8.png
[3]: /images/20221020/5999df6612ad434ca31e8c1b2ee091d5.png
[4]: /images/20221020/87a99d04d4e84e3eb9819cc482423e87.png
[5]: /images/20221020/9a0966c8f7fd44fb9310dd8b865b0c7e.png
[6 Inner Join]: /images/20221020/ef5765c2000c40a7ba1b5f7b0d844d90.png
[7]: /images/20221020/84de5c584a464e3d96e78f90bf845638.png
[8]: /images/20221020/9e840f3a2cba4a7fb7563bd09c139a35.png
[9 _ Processing Time]: /images/20221020/1331c0020fb7408baa447633ec746f8e.png
[10 _ Event Time]: /images/20221020/c80934d97f8c411384a8a719f830a974.png
[11 _ Event Time]: /images/20221020/8d323a6f7c5f4b08a551480c7ea7a9a9.png