流处理

流处理：数据一条一条地进入系统，以更频繁的运行处理，例如在每秒的末尾，或者当事件发生时立即处理。“流”指随时间推移逐渐可用的数据。这个概念出现在很多地方：Unix的stdin和stdout、文件系统API、TCP连接、互联网传送音视频等。本章中，我们将事件流（event stream） 视为一种数据管理机制：无界限、增量处理。

1 传递事件流

当输入是一个文件（字节序列），首先要将其解析为一系列记录。流处理的上下文中，记录通常称为事件（event） ，本质是：一个小的，自包含的，不可变的对象，包含某个时间点发生的事件的细节。一个事件通常包含一个来自时钟的时间戳，以指明事件发生的时间。

流处理中，一个事件由生产者（producer） （也称发布者或发送者）生成一次，可能由多个 消费者（consumer） （ 订阅者或接收者 ）进行处理。相关事件通常被聚合为一个 主题（topic） 或 流（stream） 。事件可能被编码为文本字符串或JSON或二进制编码。这允许我们将其附加到一个文件、插入关系表、写入文档数据库，也能通过网络将事件发送到另一个节点进行处理。

原则上讲，文件或数据库就足以连接生产者和消费者：生产者将其生成的每个事件写入数据存储，且每个消费者定期轮询数据存储，检查出新事件。当我们想要进行低延迟连续处理时，若数据存储不是为这种用途专门设计的，那么轮询开销会很大。轮询越频繁，能返回新事件的请求比例就越低，额外开销也越高。所以，最好能在新事件出现时直接通知消费者。

传统数据库做不好这些事，关系型数据库的触发器可用对变化做出反应，但是功能非常有限。现在已经有专门的工具来提供事件通知。

1.1 消息系统

消息传递系统（messaging system） ：用于向消费者通知新事件。

Unix管道或TCP连接这样的直接信道，是实现消息传递系统的简单方法。但是它们只能让一个发送者和一个接收者连接，而消息传递系统允许多个生产者将节点消息发送到同一个主题，并允许多个消费者节点接收主题中的消息。

在发布/订阅模式中，不同系统采取了不同的方法，没有通用答案。可以用以下两个问题区分这些系统：

1. 发送消息的速度处理的速度快，一般只能：

   • 丢掉消息
   • 将消息放入缓冲队列
   • 使用背压（backpressure） （也称为流量控制，阻塞生产者以免发送更多消息）

2. 如果节点崩溃或暂时脱机，会发生什么情况？ —— 是否会有消息丢失？

直接从生产者传递给消费者

许多消息传递系统使用生产者和消费者之间的直接网络通信，而不通过中间节点，例如UPD和无代理消息库，还有StatsD等使用UDP监控网络中机器的，还有webhooks：当消费者在网络上公开了服务，生产者可以直接发送HTTP或RPC请求将消息推送给使用者（参阅“REST和RPC”）。

这些消息传递系统在设计它们的环境中运行良好，但它们通常要求代码意识到消息丢失的可能性。它们容错程度极为有限：无法知道生产者和消费者是否还在线。

若消费者脱机，则消息会丢失。生产者崩溃时，会丢失消息缓冲区及其本该发送的消息。

消息队列

消息代理（message broker） 也称消息队列（message queue） ，实质上是一种针对处理消息流而优化的数据库。它作为服务器运行，生产者和消费者作为客户端连接到服务器。生产者将消息写入代理，消费者通过从代理那里读取来接收消息。

通过将数据集中在代理上，将持久性问题转移到代理身上，系统可以容忍客户端的频繁断开与连接。消息是否保存在内存中取决于配置，写入磁盘可以让代理在崩溃时不会丢失。针对缓慢的消费者，默认设置下代理一般会允许无上限的排队。

排队会导致异步（asynchronous） ：生产者发到代理就不管了，不会等消息被消费者处理。

消息代理与数据库对比

• 同：有些消息代理能用XA或JTA参与两阶段提交协议。

• 异：

  • 消息代理会在递给消费者时自动删除消息，而数据库会保留至显示删除。
  • 由于它们很快就能删除消息，所以大多数代理假设它们的队列很短。若消息处理时间变长，整体吞吐量会恶化。
  • 数据库通常支持二级索引和搜索数据，而消息代理只支持按照某种模式匹配主题，订阅其子集。
  • 查询数据库时，结果通常基于某个时间点的数据快照。而消息代理不支持查询，但是当数据发生变化时（即新消息可用时），它们会通知客户端。

多个消费者

多个消费者从同一主题中读取消息时，有使用两种主要的消息传递模式

• 负载均衡（load balance）：每条消息都被传给消费者之一。工作能被多个消费者共享，代理可以为消费者任意分配消息。用于并行处理消息。
• 扇出（fan-out）：每条消息被传给所有消费者。扇出允许消费者收听相同广播而不相互影响。

两种模式可以组合使用：例如，两个独立消费者组可以各订阅一个主题，每个组都共同收到所有消息，但每一组内部，每条消息仅由单个节点处理。

确认与重新支付

消费随时可能崩溃，代理向消费者发送消息，消费者可能因崩溃而没有处理或只做了部分处理。消息代理使用确认来确保消息不丢失：客户端处理完毕时通知代理，然后代理才将消息从队列中移除。若与连接关闭或超时，代理会将消息递送给另一个消费者。（会有特殊情况：确认在网络中丢失，则需要一种原子提交协议）

当与负载均衡相结合时，重传会对消息的顺序发生影响。如图，未确认消息m3随后被发给消费者1，于是消费者1按m4、m3、m5的顺序处理消息，即交付顺序与发送顺序不同。

负载均衡不可避免的打乱消息排序，若想按顺序，可以让每个消费者使用单独队列。

1.2 分区日志

不会建立日志：通过网络发送数据包、发送请求，消息代理。

会建立日志：数据库、文件系统。

既有数据库的持久存储，又能有消息传递的低延迟通知：基于日志的消息代理（log-based message brokers） 。

使用日志进行消息存储

日志只是磁盘上简单的仅追加记录序列（参见第三章-日志结构... 、第五章-复制的上下文）。

这个结构可以用于实现消息代理：生产者通过将消息追加到日志末尾来发送消息，而消费者通过依次读取日志来接收消息。若消费者读到日志末尾，则会等待新消息追加。Unix工具tail -f 能监视文件被追加写入的数据，基本上就是这样工作的。

为提高吞吐量可以对日志进行分区：如图所示，不同分区托管在不同的机器上，且每个分区都拆分出一份独立于其他分区进行读写的日志。一个主题可以定义为一组携带相同类型消息的分区。

每个分区内，代理为每个消息分配单调递增的序列号或偏移量（offset） 。分区是追加写入，所以分区内消息完全有序。但是没有跨不同分区的顺序保证。

Apache Kafka就是基于日志的消息大力。每秒百万条消息吞吐量，通过复制实现容错。

日志与传统消息相比

基于日志的消息代理天然支持扇出，因为多个消费者可以独立读取日志且不相互影响。负载均衡时，代理可以将整个分区分配给某个消费者。

每个客户端消费指派分区中所有消息，然后分配分区中的所有消息。用户会序读取被指派的分区中的消息。这种粗粒度的负载均衡有一些缺点：

• 共享消费主题工作的节点数，最多为该主题中的日志分区数，因为一个分区对应一个节点。
• 若某条消息处理得慢，会阻塞后续消息。

我们可以得出结论：

• 当消息处理代价高昂、需逐条并行处理、顺序不重要时：使用JMS/AMQP风格。
• 当消息吞吐量高、处理迅速、顺序重要时：使用基于日志的方法。

消费者偏移量

顺序消费可以很简单的判断消息是否被处理：所有偏移量小于消费者的当前偏移量的消息已经被处理。因此，代理不需要跟踪确认每条消息，而只需要定期记录消费者的偏移，这会减少开销和提高吞吐量。

这种偏移量类似于单领导者数据库复制中的日志序列号（参见“设置新从库”）：日志序列号允许跟随着断开重连后，不跳过任何写入的情况下恢复复制。原理完全相同——消息代理像个主库，消费者则是从库。

若消费者节点失效，则失效消费者分区将指派给其他节点，并从最后记录的偏移量开始消费消息。但是：当消费者1处理完但是没记录偏移量，那么重启后会发现消费者2也处理了一次，那么消息被处理了两次。

磁盘空间使用

若只追加写入日志，则磁盘空间终究耗尽。日志实际上被分割成段，并不时将旧段删除或移动到归档存储。那么，日志可以看作一个在磁盘上的大缓冲区，缓冲区填满时丢弃旧消息——也被称为循环缓冲区（circular buffer） 或环形缓冲区（ring buffer） 。实践中的部署很少能用满磁盘的写入带宽，所以通常可以保存几天甚至几周的日志缓冲区。

吞吐量对比：

• 基于日志的消息代理：吞吐量基本保持不变，因为无论如何消息都会被写入磁盘。
• 将消息保存在内存中：队列很短则系统很快，开始写入磁盘时变得很慢。

当消费者跟不上生产者

若一个消费者太慢，以至于消费偏移量指向了被删除的段，那么它会错过一些消息。而日志就是“缓冲”的一种形式，可以解决“跟不上”的问题。

但若消费者远远落后，以至于磁盘将消息删除的话，我们可以监控消费者落后日志头部的距离，若落后太多就发出报警。由于缓冲区很大，因此有足够时间等到运维人员来。

即使消费者真的开始丢失消息，也不会影响其他消费者，这有利于运维：可以实验性地消费生产日志以进行开发、测试、调试，而不担心中断生产服务。消费者关闭崩溃时会停止消耗资源，只剩下消费偏移量。这与传统消息代理不同，传统的话需要小心删除消费者已经关闭的队列，否则队列就会积累不必要的信息，与其他活着的消费者抢占内存。

重播旧信息

• AMQP和JMS风格消息代理：处理和确认消息是破坏性操作，会导致消息在代理上被删除。
• 基于日志的消息代理：使用消息是从文件中读数据，只读而不更改日志。

唯一的副作用是增加消费者偏移量，但是偏移量在消费者控制下可以很容易操纵：我们可以用昨天的偏移量跑一个消费者副本。这方面很像上一章的批处理，衍生数据通过可重复的转换过程与输入数据显示分离。它允许进行更多实验，更容易从错误和漏洞中恢复，使其成为在组织内集成数据流的良好工具。

2 流与数据库

可以从消息传递和流中获取灵感，并应用于数据库。

事件是某个时刻发生的事情的记录，而事实上，复制日志可以看作 数据库写入事件 的流：主库在处理事务时生成，而从库将写入流应用到它们自己的数据库副本。日志中的事件描述发生的数据更改。

本节中，我们讨论如何通过事件流的想法解决异构数据系统中的一个问题。

2.1 保持系统同步

我们知道，实践中通常组合几种不同存储技术来满足所有需求。由于相同或相关数据出现在不同地方，因此相互间要保持同步：如果某个项目在数据库中被更新，它也应当在缓存，搜索索引和数据仓库中被更新。

对于数据仓库，通常用ETL进程执行同步（也就是批处理）。但是，若周期性的完整数据库转储过于缓慢，有时会使用双写（dual write） 来代替。代码在数据变更时明确写入每个系统：例如，先写入数据库，然后更新搜索索引，然后使缓存项失效（甚至同时执行这些写入）。

然而，双写会导致并发问题，例如下图：俩客户端同时更新X，它们先将新值写入数据库，然后再写入搜索索引。运气不好的是——这些请求的时序是交错的，这导致了两个系统的永久不一致。

除非有额外的检测并发写入的机制，否则甚至不会意识到发生了并发写入。双写时，还有可能两个写入没有同时成功或失败，这是一个容错问题而不是并发问题，需要原子提交但是代价昂贵。

如果改成单领导者模式——例如数据库是领导者，搜索索引是从库，那么情况会好很多。这在实践中能否实现？

2.2 变更数据捕获

几十年来，数据库根本就没有能获取它日志的方式。复制日志一直被当做数据库的内部实现细节，而非公开的API，即客户端无法通过解析复制日志来提取数据。

于是，变更数据捕获（change data capture, CDC） 诞生了：观察并提取数据库的数据变更，然后将变更转换为可以复制到其他系统中形式。如图，将数据按顺序写入一个数据库，然后按相同顺序将这些变更应用到其他系统。

变更数据捕获的实现

可以将日志消费者称为衍生数据系统，即存在搜索索引和数据仓库中的数据，只是记录系统的额外视图。变更数据捕获（CDC）能确保对记录系统做的所有更改都反映在衍生数据系统中。本质上，CDC使得被捕获变化的数据库成为领导者，而基于日志的消息代理适合从源数据库传输变更事件，因为它保留了消息的顺序。

CDC通常是异步的：记录数据库系统不会等待消费者应用变更再进行提交。优点是，添加缓慢的消费者不会过度影响记录系统。缺点是，能产生所有复制延迟问题。

初始快照

如果我们有所有变更日志，那么重放日志可以重建数据库的完整状态。但是日志保留费空间，重放又费时间，所以日志需要被截断。数据库快照必须与日志变更中的偏移量相对应，才能知道从哪里开始应用变更。

日志压缩

若只能保留有限的历史日志，那么每次日志更新都要做一次快照。日志压缩（log compaction） 能解决这个问题。我们在日志结构存储引擎的上下文中讨论了“Hash索引”中的日志压缩，原理是定期在日志中查找具有相同键的记录，丢掉重复内容，只保留每个键的最新更新。

日志结构存储引擎中，NULL（墓碑（tombstone） ）的更新表示改键被删除，它会在日志压缩过程中被移除。但只要键不被覆盖或删除，它就会永远留在日志中。这种压缩日志所需的磁盘空间仅取决于数据库的当前内容——之前的值会被覆盖。

这个想法也适用于基于日志的消息代理，以及变更数据捕获的上下文。我们可以使用它来获取数据库的完整副本，而无需从CDC源数据库取一个快照。Kafka支持这种日志压缩功能，它允许消息代理被当成持久性存储使用而不是临时消息。

变更流的API支持

数据库开始将变更流作为第一类接口，而不是费工夫逆向工程一个CDC。

例如RethinkDB允许订阅通知，VoltDB允许以流的形式连续从数据库中到处数据，Kafka Connect将CDC与Kafka集成。

2.3 事件溯源

事件溯源（ Event Sourcing） 包含了一些关于流处理系统的有用想法。这是一个诞生于 领域驱动设计（domain-driven design, DDD） 社区中的技术。

事件溯源将所有对应用状态的变更存储为变更事件日志。与变更数据捕获（CDC）最大的区别就是将这一想法应哟到了不同抽象层次上：

• CDC中，应用以可变方式（mutable way） 使用数据库，任意更新和删除记录。变更日志从数据库底层获取，以确保日志中的写入顺序是对的。写入数据库的应用不需要知道CDC的存在。
• 在事件溯源中，应用逻辑显式构建在写入事件日志的不可变事件上。在这种情况下，事件存储是仅追加写入的，更新与删除是不鼓励的或禁止的。事件被设计为旨在反映应用层面发生的事情，而不是底层的状态变更。

事件源将用户行为记录为不可变的事件，而不是在可变数据库中记录这些行为的影响。事件代理使得应用随时间演化更为容易，通过事实更容易理解事情发生的原因，这利于调试和防止应用Bug。

事件溯源类似于编年史（chronicle） 数据模型，事件日志与星型模式中的事实表之间也存在相似之处。

从事件日志中派生出当前状态

事件日志本身没什么用，因为用户不需要变更历史。因此，使用事件溯源的应用需要拉取事件日志，然后转换为适合向用户显示的应用状态。转换必须是确定的，以便再次运行能产生相同的应用状态。

与CDC一样，重放时间日志允许重新构建系统当前状态。不过，日志压缩需要采用不同方式处理：

• 用于记录更新的CDC事件通常包含记录的完整新版本，因此主键的当前值完全由该主键的最近事件确定，而日志压缩可以丢弃相同主键的先前事件。
• 事件溯源在更高层次进行建模：事件通常表示用户操作的意图，而不是因为操作而发生的状态更新机制。

使用事件溯源的应用通常有些机制，能从事件日志中导出当前状态快照，因此它们不需要重复处理完整日志，这可以加速读取，提高崩溃恢复速度。

命令与事件

事件溯源的哲学是仔细区分事件（event） 和命令（command） 。来自用户的请求到达时，一开始是一个命令，应用得先验证它是否可以执行该命令，验证成功才变为一个持久化不可变的事件。

事件生成的时刻，它就成为了事实（fact） 。事件流的消费者不允许拒绝事件：当消费者看到事件时，它已经是日志中不可变的一部分。因此，对任何命令的验证，都需要在它成为事件之前同步完成。例如通过一个可自动验证命令的可序列化事务来发布事件。

状态，流和不变性

不变性原则使得事件溯源与变更数据捕获十分强大。我们通常将数据库视为应用程序当前状态的存储，状态是会变化的，那这又是如何符合不变性的呢？

只要你的状态发生了变化，那么这个状态就是这段时间中事件修改的结果。那么，是一系列不可变的事件导致了状态的变化，因此可变状态与不可变事件的仅追加日志之间并不矛盾。变化日志（change log） ，表示了随时间演变的状态。

用数学表示，应用状态是事件流对时间求积分得到的结果，而变更流是状态对时间求微分的结果。如图

如果你持久存储了变更日志，那么重现状态就非常简单。如果你认为事件日志是你的记录系统，而所有的衍生状态都从它派生而来，那么系统中的数据流动就容易理解的多。

正如帕特·赫兰（Pat Helland）所说：事务日志记录了数据库的所有变更。高速追加是更改日志的唯一方法。从这个角度来看，数据库的内容其实是日志中记录最新值的缓存。日志才是真相，数据库是日志子集的缓存，这一缓存子集恰好来自日志中每条记录与索引值的最新值。

而日志压缩是连接日志与数据库状态之间的桥梁：只保留每条记录的最新版本，丢弃被覆盖的版本。

不可变事件的优点

可审计性在金融系统中尤其重要，会计当然不能修改以前资金。

错误代码破坏数据库时，使用不可变的仅追加日志可以很容易的进行故障恢复。

从同一事件日志中派生多个视图

可以针对不同的读取方式，从相同的事件日志中衍生出不同的表现形式。效果就像一个流的多个消费者一样。例如，分析型数据库Druid用这种方式直接从Kafka摄取数据，Pistachio是一个分布式的键值存储，使用Kafka作为提交日志，Kafka Connect能将来自Kafka的数据导出到各种不同的数据库与索引。这对于其他存储系统、索引系统、从分布式日志中获取输入 来说十分重要。

添加从事件日志到数据库的显示转换，能使应用更容易随时间演进：可以使用事件日志构建一个单独的，针对新功能的读取优化视图，无需修改现有系统而与之共存。并行运行新旧系统通常比模式迁移更容易：当我们不需要旧系统时直接关闭并回收资源即可。

数据库都希望支持某些特定查询和访问模式，这导致很多模式设计，索引和存储引擎的许多复杂性。因此，将数据的读写形式分离，并允许几个不同的读取视图，可以获得很大的灵活性。这就是命令查询责任分离（command query responsibility segregation, CQRS） 。此外，针对读取进行优化时，“读取优化的视图中的数据”可以和“写入数据库时的数据”的形式不同，因为翻译过程有使其与事件日志保持一致的机制。

并发控制

事件溯源和变更数据捕获的最大缺点是：事件日志的消费者通常是异步的，这可能导致”读己之写“的问题（参见“读己之写”）。例：用户写入日志，然后从日志衍生视图中读取，结果发现他的写入还没有反映在读取视图中。

解决方案：将事件附加到日志时同步执行读取视图的更新。将这些写入操作合并为一个原子单元需要事务，所以只能将事件日志和读取视图保存在同一个存储系统中，或是跨不同系统进行分布式事务，亦或是使用全序广播（参见“全序广播实现线性化存储”）。

从事件日志导出当前状态简化了并发控制的某些部分。许多对于多对象事务的需求（参阅“单对象和多对象操作”）源于单个用户操作需要在多个不同的位置更改数据。通过事件溯源，你可以设计一个自包含的事件以表示一个用户操作。然后用户操作就只需要在一个地方执行单次写入操作——即将事件附加到日志中——这是很容易原子化的。

如果事件日志与应用状态以相同的方式分区（例如，处理分区3中的客户事件只需要更新分区3中的应用状态），那么直接使用单线程日志消费者就不需要写入并发控制了，因为它从设计上一次只处理一个事件（参阅“真的的串行执行”）。日志通过在分区中定义事件的序列顺序，消除了并发性的不确定性。如果一个事件触及多个状态分区，那么需要做更多的工作，我们将在第12章讨论。

不变性的限制

许多不使用事件溯源模型的系统也依赖不可变性：数据库中用来支持时间点快照，Git也是依靠不可变数据来保存版本历史记录。

能否永远保持所有变更的不变历史，取决于数据集的流失率：如果更新/删除率高的话，不可变的历史可能增至难以接受的巨大，碎片化就会成为问题，这时就需要压缩与垃圾收集。

除性能外，还有管理等方面因素需要删除数据，例如隐私条例要求自动删除个人信息。而在这种情况下，只把数据标记为删除是不够的——我们是想假装数据一开始没有写入。例如，Datomic管这个特性叫切除（excision） ，而Fossil版本控制系统有一个类似的概念叫避免（shunning） 。

真正删除数据非常困难，因为副本存在于很多地方：存储引擎、文件系统、SSD通常会向一个新位置写入，而不是原地覆盖旧数据，而备份通常是不可变的，防止意外删除或损坏。删除更多的是“使取回数据更困难”，而不是“使取回数据不可能”。

3 流处理

目前为止，本章讨论了流的来源和流如何传输。这一节来讨论我们可以用流做什么，即怎么处理它。一般来说有三种选项：

1. 将事件中的数据写入存储系统（数据库、缓存、搜索索引等）。这能很好的让数据库与系统的其他部分保持同步。
2. 将事件推送给客户，或将事件流式传输到可实时显示的仪表板上。这种情况下，人是流的最终消费者。
3. 可以处理一个或多个输入流，并产生一个或多个输出流。流可能经过以上两个过程组成的流水线，最后再输出。

本章的剩余部分将讨论上述选项3：处理流以产生其他衍生流。处理这样的流的代码成为算子（operator） 或作业（job） 。它与MapReduce模式相似：输入流只读，输出流仅追加；使用分区和并行化模式；基本的Map操作（转换、过滤）。

流处理与批处理相比，一个关键的区别是：流不会结束。这会带来很多差异：

• 无法排序：无法使用排序合并联接。
• 容错机制改变：无法重跑作业。

3.1 流处理的应用

流处理长期被用于监控。例如：

• 欺诈检测系统，确定信用卡、账号使用模式是否意外变化，检测盗刷盗号。
• 交易系统检查金融市场的价格变化，根据指定的规则进行交易。
• 制造系统监控工厂中机器的状态。
• 军事和情报系统跟踪潜在侵略者的活动，袭击征兆时发出警报。

随着时代进步，流处理的其他用途开始出现。

复合事件处理

复合事件处理（complex, event processing, CEP） 允许指定规则以再流中搜索某些事件模式。用于分析事件流，尤其适用于需要搜索某些事件模式的应用。

CEP系统通常用更高层次的声明式查询语言（比如SQL）或图形用户界面来描述应该检测到的事件模式。引擎在内部维护一个执行所需匹配的状态机，发现匹配时发出一个复合事件（complex event） ，并附有检测到的事件模式详情。

CEP中，查询和数据之间的关系与普通数据库相比是颠倒的：通常情况下，数据库持久存储数据并临时进行查询，而CEP引擎是长期查询的，来自输入流的事件不断流过它们，搜索匹配事件模式的查询。

流分析

CEP与流分析之间的边界是模糊的，分析一般不关注找出特定事件序列，而是关注大量事件上的聚合与统计指标：

• 测量某种类型事件的速率（每个时间间隔内发生的频率）
• 滚动计算一段时间窗口内某个值的平均值
• 将当前的统计值与先前的时间区间的值对比（例如，检测趋势，当指标与上周同比异常偏高或偏低时报警）

一般在固定时间区间内计算，几分钟内区平均，能抹平秒和秒之间的无关波动，且仍然能向你展示流量模式的时间图景。聚合的时间间隔称为窗口（window） ，将在接下来详细讨论。

有时会使用概率算法进行优化，减少内存使用。同时这不会损失精确：流处理没有任何内在的近似性。

许多开源分布式流处理框架针对分析设计：Apache Storm、Spark Streaming等。

维护物化视图

前面提到，数据库的变更流可以用于维护衍生数据系统（如缓存，搜索索引和数据仓库），使其与源数据库保持最新（参见”数据库和数据流“）。我们可以将这些示例视作维护物化视图（materialized view） 的一种具体场景（参见”聚合:数据立方体和物化视图“）：在数据集上衍生一个视图以便高效查询，当底层数据变更时更新视图。

事件溯源中，应用程序的状态是通过应用（apply） 事件日志来维护的，这里的应用状态也是一种物化视图。与CEP不同的是，只考虑某个时间窗口内的事件是不够的，构建物化视图需要所有事件，需要一个一直延伸到时间开端的窗口。

原则上讲，任何流处理组件都能用于维护物化视图，尽管“永远运行”与一些面向分析的框架假设的“主要在有限时间段窗口上运行”背道而驰。Kafka Streams支持这种用法，建立在Kafka对日志压缩comp的支持上。

在流上搜索

除了允许搜索由多个事件构成模式的CEP外，有时也存在基于复杂标准（例如全文搜索查询）来搜索单个事件的需求。

例如，媒体监测服务可以订阅新闻、搜索新闻。原理是先构建一个搜索查询，然后不断将新闻项的流与该查询进行匹配。

传统的搜索引擎先索引文件，再在索引文件上跑查询。而搜索数据流则相反（与CEP相似）：查询被存储下来，文档从查询中流过。

消息传递和RPC

消息传递系统可以作为RPC的替代方案（参见“消息传递数据流”），即作为一种服务间通信的机制，就像在Actor模型中使用的那样。尽管这些RPC类系统也基于消息和事件，且与流处理间有交叉领域，但通常不视作流处理组件：

• Actor框架主要是管理模块通信的并发和分布式执行的一种机制，而流处理主要是一种数据管理技术。
• Actor之间的交流往往是短暂的，一对一的；而事件日志则是持久的，多订阅者的。
• Actor可以以任意方式进行通信（允许包括循环的请求/响应），但流处理通常配置在无环流水线中，其中每个流都是一个特定作业的输出，由良好定义的输入流中派生而来。

也可以用Actor框架处理流。但是它的容错很低，崩溃时不能保证消息传递，除非实现了额外的重试逻辑。Apache Storm有分布式RPC功能，允许用户查询分散到一系列也是处理事件流的节点上，这些查询与来自输入流的事件交织，将结果汇总并发回给用户。

3.2 时间推理

流处理通常需要和时间打交道，尤其是用于分析目的的时候，会频繁使用时间窗口。而”最后五分钟“的含义其实是非常棘手的。

• 批处理中，大量历史时间迅速收缩，可以在几分钟内读取一年的历史时间。批处理检查每个事件中嵌入的时间戳，时间戳固定使得使得处理是确定性的，相同的输入产生相同的结果。

• 流处理中，许多框架使用本地系统时钟（处理时间（processing time） ）来确定窗口。这种方法的优点是简单，事件创建与事件处理之间的延迟可以忽略不计。然而，当有任何显著的处理延迟时，处理就失效了。

事件时间与处理时间

很多原因都可能导致处理延迟：排队，网络故障（参阅“不可靠的网络”），性能问题导致消息代理/消息处理器出现争用，流消费者重启，重新处理过去的事件（参阅“重放旧消息”），或者在修复代码BUG之后从故障中恢复。消息延迟还会导致无法预测消息顺序，流处理算法需要专门编写，以适应这种时机与顺序的问题。

将事件时间和处理时间搞混会导致错误的数据。当重新部署流处理器时，流处理器会停止一小段时间，并在恢复后处理积压时间。如果按照处理时间来衡量速率，那么在处理积压日志时，请求速率看上去就像有一个异常的突发尖峰，而实际上请求速率是稳定的

知道什么时候准备好了

用事件时间定义窗口，会导致永远无法确定是否受到特定窗口的所有事件（是否还有事件在来的路上）。例如，将事件分组为1分钟的接口以统计每分钟的请求数，就算现在进入的主要都是第二分钟和第三分钟的事件，我们也无法确定第一分钟的事件全部收到了。

我们需要能处理这种在窗口宣告完成后到达的滞留（straggler） 事件，大致有两种选择：

1. 忽略滞留事件，因为正常情况下它们只是事件中的一小部分。可以将丢弃事件的数量作为一个监控指标，当大量丢消息时报警。
2. 发布一个更正（correction） ，一个包括滞留事件的更新窗口值。更新的窗口与包含散兵队员的价值。你可能还需要收回以前的输出。

你用的是谁的时钟？

当事件可能在系统内多个地方缓冲时，为事件分配时间戳变得困难。应用可能在脱机时使用，重新连上网时才上报所有事件，对于这个流的消费者来说来说，它们像是延迟极大的滞留事件。这种情况下：

• 有意义的时间戳应该是设备上用户交互的时间，但是用户的时钟通常是不可信的（参照”时钟同步与准确性“）。
• 服务器时钟在描述用户交互方面意义不大。

要校准不正确的设备时钟，一种方法是记录三个时间戳：

• 事件发生的时间，取决于设备时钟
• 事件发送往服务器的时间，取决于设备时钟
• 事件被服务器接收的时间，取决于服务器时钟

通过从第三个时间戳中减去第二个时间戳，可以估算设备时钟和服务器时钟之间的偏移，然后将其应用于事件时间戳，从而估计事件实际发生时间。

批处理也有这样的时间问题，但是在流处理的上下文中，我们更容易意识到时间的流逝。

窗口的类型

• 滚动窗口（Tumbling Window）：固定长度，每个事件只能属于一个窗口。例如，一个一分钟的滚动窗口，第一个在0:1:00-0:1:59，第二个在0:2:00-0:2:59。
• 跳动窗口（Hopping Window）：固定长度，允许窗口重叠以提供一些平滑。例如，一个跳跃1分钟步长5分钟的窗口，第一个窗口覆盖0:1:00-0:4:59分，第二个覆盖2:00-6:59分。
• 滑动窗口（Sliding Window）：包含彼此间距在特定时长内的所有事件，边界是移动的。例如，有一个5分钟的滑动窗口，随着时间推进，窗口内5分钟之前的事件会被不断移除，而新事件不断加入。
• 会话窗口（Session Window）：没有固定持续时间。将同一用户出现时间相近的所有事件分组在一起，当用户一段时间没有活动时窗口结束。会话切分是网站分析的常见需求。

3.3流式连接

流流连接（窗口连接）

假设网站要找出搜索URL的趋势。进行搜索时，记录这个包含查询及其返回结果的事件。每当有人点击一个搜索结果时，记录另一个点击事件。为了计算搜索结果中每个URL的点击率，需要将搜索动作与点击动作的事件的两个流联接在一起，这些事件通过相同的会话ID进行连接。

可能会产生以下情况：

• 用户丢弃了搜索结果（啥都没点就把网页关了）：点击事件则永远不会发生。

• 搜索结果没有被丢弃：

  • 搜索与点击之间的时间是高度可变的：一般是几分钟内，但也可能长达几天（即用户没关网页 ，一段时间后重新回到浏览器页面上，并点击了一个结果）。
  • 网络延迟：点击事件可能比搜索事件先到达。这时，可以选择合适的窗口，例如连接点击和搜索间隔一小时内的事件。

为了实现这种链接，流处理器需要维护状态：例如，按会话ID搜索最近一小时内发生的所有事件。无论何时发生搜索事件或点击事件，都会被添加到合适的索引中，而流处理器也会检查另一个索引是否有具有相同会话ID的事件到达。若有匹配事件就会发出一个表示搜索结果被点击的事件；如果搜索事件直到过期都没看见有匹配的点击事件，就会发出一个表示搜索结果未被点击的事件。

流表连接（流扩展）

可以用数据库的信息来扩充（enriching） 活动事件。例如流入用户ID，输出的时候将用户ID扩展为用户的档案信息。为了执行此联接，流处理器应为每个活动事件在数据库中查找其对应ID，然后将获得到的用户信息添加到活动事件中。

• 通过查询远程数据库来实现，但可能很慢甚至导致数据库过载。
• 将数据库副本加载到流处理器中本地查询。这与在”Map端连接“中讨论的散列连接十分相似：如果数据库本地副本足够小，则可以是内存中的散列表，比较大的话也可以是本地磁盘上的索引。

与批处理相比：批处理作业使用数据库的时间点快照作为输入，而流处理器长时间运行且数据库内容随时间而改变，所以流处理器数据库的本地副本需要保持更新。这可以通过变更数据捕获来解决档案更新，因此我们有了两个流的连接：活动事件和档案更新。

流表连接与流流连接最大的区别在于——对于表的变更日志流，连接了一个可以回溯到”时间起点“的窗口（概念上是无限的窗口），新版本的记录会覆盖更早的版本。

表表连接（维护物化视图）

在”描述负载“中说过，用户要查看他们主页时间线时，迭代用户所关注人群的推文合并它们是一个开销巨大的操作。我们可以用一个时间线缓存：一种每个用户的“收件箱”，在发送推文的时候写入，读取时间线时简单地查询即可。物化与维护这个缓存需要处理以下事件：

• 用户发送推文时，将推文添加到每个关注该用户的时间线上。
• 用户删除推文时，从所有用户的时间表中删除。
• 用户a关注b时，a最近的推文添加到b的时间线上。
• 用户a取消关注b时，a将推文从b的时间线中删除。

这需要两个事件流：推文事件流、关注事件流。维护一个数据库：包含每个用户的粉丝集合，以便知道当一个新推文到达时，需要更新哪些时间线。

可以说，这个流处理维护了一个连接了这两个表（推文与关注）的物化视图，时间线其实是这个视图的缓存，每当基础表发生变化时都会更新。这个物化连接的变化甚至遵循乘积法则，将一个连接看成u*v，会发现 (u·v)’= u’v + uv’(u·v)’= u’v + uv’ 。任何推文的变化量都与当前的关注联系在一起，任何关注的变化量都与当前的推文相连接。

连接的时间依赖性

这三种连接有很多共通之处：都需要流处理器维护连接一侧的一些状态，当连接另一侧的消息到达时查询该状态。

时许依赖很重要——用于维护状态的事件顺序会出现在很多地方。而分区日志中，单个分区内的事件顺序是保留下来的，但跨分区就不一定。这就会产生问题：如果不同流中的几个事件几乎同时发生，应该按照什么顺序处理？流表连接中用户档案更新时，哪些应该连接新档案？

由于跨越流的事件顺序未定，连接是不确定的：输入上次相同的作业不一定得到相同的结果。在数据仓库中，这个问题被称为缓慢变化的维度（slowly changing dimension, SCD） ，通常通过对特定版本的记录使用唯一的标识符来解决。这使得连接变为确定，但会导致日志压缩无法进行：表中所有的记录版本都需要保留。

3.4 容错

流处理是如何处理容错的？第10章我们提到，批处理容错很强：MapReduce作业中任务失败，很简单地在另一台机器上再次启动，并丢弃失败任务的输出。因为输入不可变，输出写入HDFS中，而输出仅在任务成功完成后可见。

流处理中出现了同样的容错问题，但处理起来没那么直观：无法处理完一个无限的流，所以无法等待某个任务完成后再使其输出可见。

微批量与存档点

将流分解成小块，并像微型批处理一样处理每个块。这种方法被称为微批次（microbatching） ，它被用于Spark Streaming。批次大小通常为1秒，因为批次更小则调度和协调的开销大，批次越大延迟越大。微批次相当于一个与批次大小相等的滚动窗口，作业需要更大的窗口时需要显式地将状态转移到下一个批次。

Apache Flink使用了存档点。若流算子崩溃，则从最近的存档点重启，并丢弃从最近的存档点到崩溃之间的所有输出。存档点会由消息流中的 壁障（barrier） 触发，类似于微批次之间的边界，但不会强制一个特定的窗口大小。

流处理框架的范围内，微批次与存档点方法提供了与批处理一样的恰好一次语义。但是，只要输出离开流处理器，框架就无法抛弃失败批次的输出。这种情况下，重启失败任务会导致外部副作用发生两次，微批量和存档点不足以阻止这一问题。

原子提交再现

为在出现故障时表现出恰好处理一次的样子，要确保事件的所有输出当且仅当成功才生效。输出包括：发送给下游算子、发送给外部消息传递系统、数据库写入、对变更算子状态、确认输入的消息。这些事情要么原子性发生，要么不发生，但它们不应该失去同步，我们在分布式事务和两阶段提交的上下文中讨论过。

幂等性

我们的目标是丢弃任何失败任务的部分输出，以便安全重试而不会生效两次。分布式事务是实现这个目标的一种方式，另一种方式是依赖幂等性（idempotence） 。

幂等，即用户对于同一操作，发起一次请求和发起多次请求的结果是一致的。幂等操作是一种实现恰好一次语义的有效方式，额外开销很小。不是幂等的操作往往也可以通过一些额外元数据做成幂等。例如Kafka消息带有一个偏移量，写入外部数据库时带上偏移量，可以判断一条更新是否执行过，避免重复执行。

当从一个处理节点故障切换到另一个节点时，可能需要进行防护（fencing） ，以防止被假死节点干扰。

失败后重建状态

任何需要状态的流处理，都必须确保失败后能恢复状态。有两种方法：

• 将状态保存在远程数据存储中，会很慢

• 在流处理器本地保存状态。流处理器从故障状态恢复时，新任务读取状态副本，恢复处理而不丢失数据。

  • Flink定期捕获算子状态的快照，写入HDFS
  • Samza和Kafka Streams将状态变更发送到具有日志压缩功能的专用Kafka主题（有点像变更数据捕获）

某些情况下甚至不需要复制状态，因为它可以从输入流重建。如果状态从相当短的窗口中聚合而成，那么能很快地重放该窗口中的输入事件。若状态是通过变更数据捕获来维护的数据库本地副本，也可以从日志压缩的变更流中重建数据库（参阅”日志压缩“）。

所有权衡取决于底层基础架构的性能特征：某些系统中，网络延迟可能小于等于磁盘访问延迟。