流式处理的一些概念 一：时间域、窗口化（翻译）

Posted 2023-04-11

参考技术A

流式数据处理目前是大数据中的一个重要部分：

什么是“流式”（streaming）?

考虑到设计良好的流式系统与任何现有的批处理引擎一样能够产生正确、一致和可重复的结果，我更喜欢将流式这个术语隔离到一个非常特定的含义：一种考虑到无限数据集而设计的数据处理引擎。（这个定义包括真正的流式传输和微批量实现）

“流式系统”（streaming system）能做什么？

长期以来，流式系统一直被归类与能提供低延迟、不准确或推理的结果的有利市场，常常与更有能力的批处理系统结合以提供最终正确的结果，例如，Lambda 架构，基本思想是在 batch system 旁运行streaming system，两者执行基本相同的计算。流式系统提供低延迟、不准确的结果，一段时间之后，批处理系统继续运行并提供正确的输出，需要构建两套体系。

Event-time 和 Processing-time

在任何数据处理系统中，通常有两个关心的时间域（time domain）

虽然不是所有的场景都关心 event time，但是有很多情况都会关心。例如：描述用户随时间的行为、大多数计费应用程序以及许多类型的异常检测。

在理想情况下，event time 和 processing time 是相等的，事件发生时立即处理。然而，现实情况并非如此，事件时间和处理时间之间的偏差不仅非零，而且常常受到底层输入源、执行引擎和硬件特性的影响，包括：

因此，对任何现实系统中的事件时间和处理时间的进度的描述，通常都会得到一些类似于下图中的理想情况和现实情况的偏差。

虚线表示理想情况，红色线表示现实情况，理想与红线之间的水平距离是处理时间和事件时间之间的偏离。这种偏差主要是由 processing pipeline 引入。

由于事件时间和处理时间之间的映射不是静态的，这意味着如果关心它们的事件时间，就不能只在 pipeline 中观察到的情况下分析数据。大多数现有系统设计用于无界数据的操作方式。为了处理无界数据集的无限特性，这些系统通常提供一些对数据加窗口（window）的概念，本质上意味着沿着时间边界将数据集切成有限块。

如果你关心正确性，并且希望在分析事件时间下的结果，就不能基于处理时间定义的时间边界。由于处理时间和事件时间之间没有一致的相关性，一些事件时间数据最终将落入错误的处理时间窗口，从而导致失去正确性。而在无界数据中，无序和可变的偏差会引发事件时间窗口的完整性，因为缺乏处理时间和事件时间之间偏差的可预测映射。

需要一些工具来描述数据的完整性，新的数据到达，旧的数据会缩回或更新，并且系统可以自动完成这些。

现在我们开始研究在有界和无界数据处理中常用的核心使用模式类型。

处理有界数据相对简单，在下图中，从左边开始，使用一个数据集（full of entropy），通过一些数据处理引擎（批处理，如 MapReduce），在右边生成一个新的结构的数据集。

批处理系统一直用于处理无界数据集。这些方法围绕着将无界数据切片成适合于批处理的有界数据集的集合。

最常见的方法是将输入数据窗口化为固定大小的一些窗口，将每一个窗口作为单独的、有界的数据源进行处理。尤其是对于像日志这样的输入源，事件会被写入结构层次的目录和文件，它们的名称编码它们所对应的窗口。

实际上，大多数系统仍然有一些完整性的问题需要处理：如果由于网络分区，一些事件在通往日志的路径上被延迟了怎么办？如果事件是全局收集的，并且必须在处理之前转移到公共路径，该怎么办？如果你的事件来自移动设备呢？这意味着某种缓解措施（sort of mitigation）可能是必要的。例如，延迟处理直到确定已经收集了所有事件，或者每当数据到达较晚时，指定窗口重新处理整个批次。

当尝试使用批处理引擎将无界数据处理为更复杂的窗口策略（如用户会话）时，这种方法会更加崩溃。会话通常被定义为由一段不活动的间隙（a gap of inactivity）终止的活动时间段（period of activity for a specific user）。当使用典型的批处理引擎计算会话时，通常以 batch split 结束会话，如下图中的红色标记所示。可以通过增加批量大小来减少 split 的数量，但是要以增加延迟为代价。另一个选项是添加额外的逻辑来拼接之前运行的会话，但代价是更加复杂。无论哪种方式，使用传统的批处理引擎来计算会话都不太理想。

与大多数基于批处理的无界数据处理方法的特殊性质相反，流式系统针对无界数据构建。对于许多真实世界的分布式输入源，不仅发现在处理无界数据，还要处理以下数据：

在处理具有这些特性的数据时，有几种方法可以采用。通常将这些方法分为四类：

用于时间基本上不相关的情况，即所有相关的逻辑都是数据驱动的。由于这些用例的所有内容都是由更多数据的到达决定的，所以流引擎除了基本数据传递之外，没有什么特别的东西需要支持。因此，基本上现有的所有流系统都支持开箱即用的、与时间无关的用例。批处理系统也非常适合于无界数据源的时间不可知处理，只需将无界数据源切成任意的有界数据集序列并独立处理这些数据集。

过滤（Filtering）

假设你正在处理 Web 流量日志，并且希望过滤掉非指定域的所有流量。每个记录到达时查看是否属于感兴趣的域，如果没有，则丢弃它。由于这类事情在任何时候都只依赖于单个元素，因此数据源是无界的、无序的、以及事件时间偏差的变化的事实是不相关的。

内连接（Inner-joins）

另一个与时间无关的例子是内部连接（或 hash 连接）。当连接两个无界数据源时，如果只关心来自两个源的元素到达时连接的结果，则逻辑中没有时间元素。当从一个源看到一个值时，可以简单地将其缓冲到持久状态；只需要在另一个源的第二个值到达时进行连接操作。

近似算法（Approximation algorithms）

类似 approximate Top-N、streaming K-means 等，它们获取无限的输入源，并提供输出数据，这些数据或多或少类似于希望得到的结果。

近似算法的优点在于，通过设计，它们是低开销的，并且是为无界数据设计的。

缺点是存在有限的算法集，算法本身通常很复杂（这使得很难产生新的算法），它们的近似性质限制了它们的实用性。

值得一提的是：这些算法设计中通常都包含一些时间元素（例如，某种内置的衰变）。由于它们在到达时处理元素，所以该时间元素通常基于处理时间。这对于在近似上提供某种可证明的误差边界的算法特别重要。如果这些错误边界是以数据按顺序到达为前提的，那么当以变化的事件时间偏移为算法提供无序数据时，它们实际上没有任何意义。

剩下的两种无界数据处理方法都是窗口化的变体。Windowing 是将数据源（无界或有界）沿着时间边界切成有限块进行处理的概念。下图显示了三种不同的窗口模式：

窗口化在所讨论的两个时间领域（处理时间和事件时间）都有意义。

处理时间窗口（Windowing by processing time）

当通过处理时间窗口化时，系统将传入数据缓冲到窗口中，直到经过了一些处理时间。例如，对于5分钟的固定窗口，系统将缓冲5分钟的处理时间，之后将把在这5分钟中观察到的所有数据作为窗口，将它们发送到下游进行处理。

处理时间窗口有几个很好的特性：

处理时间窗口有一个很大的缺点：如果所讨论的数据具有与其相关联的事件时间，那么如果处理时间窗口要反映这些事件实际发生时的真实情况，则这些数据必须按事件时间顺序到达。不幸的是，基于事件时间有序数据在许多真实的分布式输入源中并不常见（原文中举了两个例子）。

事件时间窗口（Windowing by event time）

当需要以有限块观察数据源，反映那些事件实际发生的时间时使用的窗口是事件时间窗口。这是窗户化的最高标准。可是，今天使用的大多数数据处理系统缺乏对它的原生支持。

上图中的实心白线表示两个感兴趣的特定数据。这两个数据到达的处理时间窗口都与他们本该属于的事件时间窗口不匹配。因此，如果对于关心事件时间的场景，这些数据被窗口化到处理时间窗口中，计算的结果是不正确的。只有正确的进入事件时间窗口的分析结果才会是正确的。

事件时间窗口化对于无限数据的另一个好处是，可以创建动态大小的窗口，例如会话（sessions），而不需要在固定窗口上生成会话进行随意的 split，例如之前提到的 batch split。

当然，强大的语义都是有代价的，事件时间窗口也不例外。事件时间窗口有两个明显的缺点，因为窗口通常必须（相比处理时间）比窗口本身的实际长度更长：

下一篇文章会更加具体的讨论上面提到的问题的细节

原文： https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101

「Flink」理解流式处理重要概念

什么是流式处理呢？

这个问题其实我们大部分时候是没有考虑过的，大多数，我们是把流式处理和实时计算放在一起来说的。我们先来了解下，什么是数据流。

数据流（事件流）

• 数据流是无边界数据集的抽象
• 我们之前接触的数据处理，大多都都是有界的。例如：处理某天的数据、某个季度的数据等
• 无界意味着数据是无限地、持续增长的
• 数据流会随着时间的推移，源源不断地加入进来
• 数据流无处不再
• 信息卡交易
• 电商购物
• 快递
• 网络交换机的流向数据
• 设备传感器发出的数据
• …
• 这些数据都是无穷无尽的
• 每一件事情，都可以看成事件序列
• 数据流是有序的
• 数据的到来总是有个先后顺序
• 数据流是不可变的
• 事件一旦发生，就不能被改变
• 它陈述了某一个时刻的事实
• 数据流是可以重播的
• 为了处理的一些问题、纠正过去的错误，可以重跑数据流
• 借助于Kafka，我们可以重新消费几个月之前的原始数据流

流式处理

流式处理就是指实时地处理一个或多个事件流。它是一种编程范式。其他编程领域，主要有3种编程范式：

1. 请求与响应
• 延迟最小的一种方式，响应时间要求亚毫秒级到毫秒之间
• 响应时间一般分稳定
• 发出请求，等待响应（大部分的JavaEE同学，都是开发这一类编程范式的应用），其实就是OLTP
2. 批处理
• 特点：高延迟、高吞吐
• 一般是固定某个时刻开始启动执行，读取所有的数据，然后输出接口
• 每次读取到的都是旧数据
• 主要应用在DWH或BI中
3. 流式处理
• 特点：介于上述两者之间
• 流式处理可以让业务报告保持更新，持续响应

流的定义不依赖某个框架，只要储蓄从一个无边界数据集中读取数据，并对它们进行处理生成结果，就是进行流式处理。重点是：整个过程必须是持续的。

流式处理中的时间

上述我们已经说过了，数据流都是有序的。某一时刻的数据是确定的。时间是流式处理中非常重要的概念。大部分流式应用的操作都是基于时间窗口的。

流式系统一般包含以下几个时间概念（熟悉Flink的同学应该会很熟悉）：

• 事件时间（Eventtime）
• 事件实际发生的时间
• 用户一般只对事件发生时间感兴趣
• 日志追加时间
• 日志追加时间是指事件保存到事件存储源的时间
• 例如：数据是什么到达Kafka的（Kafka是可以启用自动添加时间戳功能的）
• 处理时间
• 流式处理应用接收到事件后，要对齐进行处理的时间
• 处理时间取决于流式处理应用何时读取到这个时间
• 如果应用程序使用了两个线程来读取同一个事件，这个时间戳可能会不一样
• 这个时间戳非常不可靠，应该避免使用它

状态

如果流式处理是来一个事件就处理一个事件，那么流式处理就很简单。但如果操作中包含了多个事件，流式处理就有意思了。例如：我们想在流式处理中统计北京用户的订单数量、消费金额等等。此时，就不能光处理单个事件了，我们需要获取更多的事件。事件与事件之间的信息就称之为状态。例如简单的，求某个类型的订单数等。

这些状态一般就保存在流式处理程序本地变量（本地内存）中，例如：使用HashMap来保存计数。但这种做法是很不可靠的，流式处理处理的是无界数据集，一旦应用程序出现异常，就会出现状态丢失，这是我们说不能接受的。所以，每一种流式计算框架都会很小心地持久化状态。如果应用程序重启，需要将这些数据恢复。

流式处理一般包含两种状态：

• 本地状态
• 这种状态只能被应用程序实例访问（不过Flink 1.9版本是可以外部来访问本地状态的）
• 内嵌到应用程序的数据库中进行维护和管理
• 特点：速度快，但受内存大小的限制，所以，很多流式处理系统都将数据拆分到多个子流中处理
• 外部状态
• 用外部存储来处理，一般使用NoSQL系统，例如：Cassadra
• 特点：没有大小限制，可以被应用程序多个实例访问、甚至外部应用访问，但引入额外的系统会造成延迟、复杂性（例如：要维护内部和外部状态一致性问题）

时间窗口

大部分针对流的操作都是基于时间窗口的。例如：计算一周内销量最好的产品。两个流的合并也是基于时间窗口的。流式系统会合并发生在相同时间段上的事件。窗口是有类型的。以下几点是我们设计窗口需要考虑的：

• 窗口的大小
• 是基于5分钟计算还是基于15分钟、甚至是一天
• 窗口越小，就能越快地发现变更，不过噪声也就越多
• 窗口越大，变更就跟平滑，不过延迟也越严重
• 窗口的移动频率（移动间隔）
• 5分钟的窗口，可以1分钟计算一次，或者每秒钟计算一次，或者每当有新事件到达时计算一次
• 如果“移动频率”与窗口大小相等，这种称为滚动窗口（tumbling window）
• 如果窗口随着每一条记录移动，这种情况称为滑动窗口（sliding window）
• 窗口的可更新时长
• 假设：计算了 00:00 – 00:05 之间的订单总数，一个小时后，又得到了一些“事件时间”是 00:02的事件（例如：因为网络通信故障，这个消息晚到了一段时间），这种情况，是否需要更新 00:00 – 00:05 这个窗口的结果呢？或者就不处理了？
• 理想情况下，可以定义一个时间段，只要在这个时间段内，事件可以被添加到对应的时间片段里。例如：如果事件处于4个小时以内，就更新，否则，就忽略掉。
• 窗口时间对齐
• 窗口可以与时间对齐，例如：5分钟的窗口如果每分钟移动一次，那么第一个分片可以是：00:00 – 00:05，第二个就是 00:01 – 00:06
• 窗口也可以不与时间对齐，例如：应用可以在任何时间启动，那么第一个分片有可能是03:17 – 03:22
• 滑动窗口永远不会与时间对齐，只要有新的记录到达，就会发生移动

下面这张图，说明了滚动窗口与滑动窗口的区别。

滚动窗口：假设窗口的大小为5分钟，这里确定的3个时间窗口

滑动窗口：假设每分钟滑动一次，那么这个时候会有5个时间窗口，计算结果会发生重叠

流式处理的设计模式

单个事件处理

这是流式处理最基本的模式。这种模式也叫：map或filter模式。经常被用来过滤无用的事件或者用于转换事件。

这种模式，应用程序读取流中的数据，修改数据，然后把事件生成到另一个流上。这一类应用程序无需在程序内部维护状态，每一个事件都是独立处理的。这种错误恢复和进行负载均衡都很容易。因为无需进行状态恢复操作。

使用本地状态

大部分流式处理应用关系如何聚合数据。特别是：基于时间窗口进行聚合。例如：找到每天最低、最高的交易价格。要实现这种操作，就需要维护流的状态。例如：我们需要将最小值、最大值保存下来，用它们与每一个新值对比。这类操作，可以通过本地状态来实现。例如：每一个分组都维护自己分组的状态。

一旦流式处理中包含了本地状态，就需要解决以下问题。

• 内存使用
• 必须要有足够的内存来保存本地状态
• 持久化
• 确保应用程序关闭时，不会丢失状态
• 例如：我们可以使用RocksDB将本地状态保存到内存里、同时持久化到磁盘上，以便重启后恢复。而且需要将本地状态的变更发送到Kafka的主题上
• 重新负载均衡
• 有时候，分区被重新分配给不同的消费者。这种情况，失去分区的实例必须把最后的状态保存下来，或得分区的实例必须要知道如何恢复到正确的状态

多阶段处理和重分区

有些时候，我们要通过所有可用的数据来获得结果。例如：要发布每天的“前10支”股票，这10支股票需要从每天的交易股票中挑选出来。如果仅仅在单个实例上处理是不够的，因为10支股票分布在多个实例上。

此种，我们分为多个阶段来处理。

1、计算每支股票当天的涨跌。这个计算可以在每个实例上执行

2、将结果写入到单个分区

3、再用一个实例找出当天的前10支股票

这一类操作就与MapReduce很像了。

使用外部查找——流和表的连接

有时候，流式处理需要将外部数据和流集成在一日。例如：外部数据中保存了一些规则、或者将完整完整地用户信息拉取到流中。

这种case最大的问题，外部查找会带来严重的延迟，一般在 5-15 ms之间，这在很多情况下是不可行的。而且，外部系统也无法承受这种额外的负载——流式处理系统每秒可以处理10-50W个事件，而数据库正常情况下每秒只能处理1W个事件，所以需要伸缩性更强的解决方案。

为了获取更好的性能和更强的伸缩性，需要将外部数据库的信息缓存到流式处理应用中。但考虑以下问题：

如何保证缓存里的数据是最新的？

如果刷新太频繁，仍然会对数据库造成很大压力，缓存也就无用了。

如果刷新不及时，那么流式处理中所用的数据就会过时。

如果能够捕捉数据库的变更事件，并形成事件流，流式处理作业就可以监听事件流，并及时更新缓存。捕捉数据库的变更事件并形成数据流，这个过程称为CDC（Change Data Capture）。例如：我们可以通过Canal来捕获MySQL数据库的变化、可以通过ogg来捕获Oracle数据库的变化

流与流的连接

有时候需要连接两个真实的事件流。要连接两个流，就是连接所有的历史事件（将两个妞中具有相同键、发生在相同时间窗口内的事件匹配起来），这种流和流的连接称为：基于时间窗口的连接（windowed-join）。连接两个流，通常包含一个滑动时间窗口。

乱序事件

不管对于流式处理、还是传统的ETL系统，处理乱序事件都是一个挑战。物联网领域经常发生乱序事件：一个移动设备断开Wifi连接几个小时，在重新连上WiFi后，将几个小时堆积的事件一并发出去。要让流式处理应用处理好这些场景，需要做到几下：

• 识别乱序事件
• 应用程序需要检查事件的时间，并将其与当前时间进行比较
• 规定一个时间段用于重排乱序事件
• 例如：3个小时以内的事件可以重排，但3个小时以外的事件就可以直接扔掉
• 具有一定时间段内重排事件的能力
• 这是流式处理应用和批处理的重要不同点
• 假设有一个每天运行的作业，一些事件在作业结束之后才到达，那么可以重新运行昨天的作业来更新
• 而在流式处理中，重新运行昨天的作业是不存在的，乱序事件和新到达的事件必须一起处理
• 具备更新结果的能力
• 如果处理的结果保存在数据库你，那么可以通过put或update对结果进行更新

重新处理

该重要模式是重新处理事件：

• 流式处理应用更新了，要使用新版本应用处理同一个事件流，生成新的结果，并比较两种版本的结果，然后某个时间点将客户端切换到新的结果流
• 现有的流式处理出现了缺陷，修复后，需要重新处理并重新计算结果

第一种情况，需要Kafka将事件流长时间地保存在可伸缩的数据存储中

• 将新版本的应用作为一个新的消费者组
• 新的版本从输入主题的第一个偏移量开始读取数据
• 检查结果流，在新版本的处理作业赶上进度时，将客户端应用程序切换到新的结果流上

第二种情况，需要应用程序回到输入流的起始位置开始处理，同时重置本地状态，还要清理之前的输出流。这种方式处理起来比较困难。建议还是使用第一种方案。

参考文献：

《Kafka全文指南》