# 12. 流处理系统 派生数据 ## 1. 流处理系统 前面讨论的批处理系统存在一个重要的假设: 输入是有界的，是已知的有限大小，所以批处理知道何时读完他们。比如MapReduce 核心的排序操作必须读取整个输入，然后才开始生成输出。 而实际上，有很多数据是无限的，而且随着时间的推移逐渐到达。数据集永远不会以任何有意义的方式"完成"。而批处理则必须人为将数据划分为固定时间段的数据库分批处理。批处理的问题是，输入的更改只会在一天之后的输出中反映出来。为了更快的响应用户，我们需要完全放弃固定的时间片，每当有事件就开始处理，这就是流处理背后的思想。 ### 1.1 发送事件流 在流处理的上下文中，记录通常被称为事件，改事件或者说对象，包含某个时间点发生的事件的细节。每个事件通常包含一个时间戳，用于指示事件发生的**墙上时间**。 事件由生产者生成一次，然后可能由多个消费者处理。相关的事件通常被组合成主题或流。为了更低的响应延时，当新事件出现时，最好通知消费者。 ### 1.2 消息系统 向消费者通知新事件的常见方法是使用消息系统。在这种发布/订阅模式中，不同的系统采用了不同的方法，为了区分这些系统，提出以下两个问题对区分很有帮助: 1. 如果生产者发送消息的速度比消费者能处理的快，会发生什么？ - 一般有三种选择: 系统丢弃消息；将消息缓存在队列中；激活背压(流量控制，即阻止生产者发送消息) - 如果消息被缓存在队列中，那么队列增长时会发生什么非常重要 - 如果内存无法容纳所有队列，系统是否会崩溃，还是消息会被写入磁盘 - 如果消息会落盘，又会如何影响消息传递系统的性能 2. 如果节点崩溃或者暂时离线，是否会有消息丢失? - 持久化需要写入磁盘或者结合复制方案，这些都是有成本的 - 如果能够接受消息丢失，那么同样的硬件上可以获得更高的吞吐量和更低的延迟 消息传递有如下几种方式: 1. 生产者与消费者之间直接消息传递 2. 消息代理 #### 生产者与消费者之间直接消息传递 这种方式通常要求应用程序意识消息丢失的可能性，它们只支持有限的容错。通常还假定生产者和消费者同时在线。诸如 RPC 和 HTTP 请求，webhooks 都是这种消息传递方式 #### 消息代理 消息代理本质上是一种针对**处理消息而优化的数据库**。将数据集中在代理上，可以更容易的适应不断变化的客户端。持久性问题被转移到代理。消息写入代理的结果也通常导致**消费以异步方式工作**。 #### 多个消费者 当多个消费者读取同一主题中的消息时，有两种主要的消息传递模式: 1. 负载均衡式: - 每一条消息都只被传递给其中一个消费者，所以消费者可以共享主题中处理消息的工作 - 通过增加消费者可以并行处理消息 2. 扇出式: - 每条消息都被传递给所有的消费者 - 独立的消费者独立接收相同的消息广播 这两种模式可以组合使用 #### 确认和重传 消费者可能在未处理完消息时就已经奔溃，为了确保消息不会丢失，消息代理使用**确认**: 客户端必须在处理完消息后显示的告诉代理，以便代理可以将其从队列中移除。 但是消息的确认/重传会导致以下两个问题: 1. 消息重复: 确认的消息可能在发送给代理的过程中丢失，导致代理重复发送和消息 2. 消息乱序: 与负载均衡结合时，重传机制不可避免的会导致消息被重新排序，为了避免这个问题，可以为每个消费者使用单独的队列(即不使用负载均衡) 注意如果消息之间存在因果依赖关系，消息的顺序就是至关重要的。 ## 2. 分区日志 消息代理是基于瞬间的消息传递思想构建的，因此尽管可以永久记录消息，但通常也不会这么做。如果将新的消费者添加到消息系统，通常它只会开始接收在它注册后发送的消息；任何之前的消息已经消失。 那为什么不把数据库的持久存储和消息传递的低延迟结合起来呢？这正是**日志消息代理**背后的思想。 ### 2.1 基于日志的消息存储 日志是磁盘上一个仅支持追加式修改记录的序列。与日志结构化存储引擎类似，我们可以使用相同结构来实现消息代理: 1. 生产者通过将消息追加到日志的末尾来发送消息 2. 消费者通过一次读取日志来接受消息 为了提升吞吐量，可以对日志进行分区，在每个分区中，代理为每个消息分配一个单调递增的序列号或偏移量。因为分区日志只能追加，所以分区内的消息是完全有序的。不同分区之间则没有顺序保证。 ![基于日志的消息生产和消费](/images/db/message_log.png) kafka、Amazon Kinesis Stream 和 Twitter DistributedLog 都是这种基于日志的消息代理系统。尽管这些消息代理将所有消息写入磁盘，但是通过多台机器进行分区，能够实现每秒数百万条消息的吞吐量，并且通过复制消息实现容错性。 因为多个消费者可以独立读取日志而不相互有影响，而且消息不会从日志中删除，所以基于日志的方法很自然的支持扇出式消息传递。通过将整个分区分配给消费者组中的节点，每一个节点消费一个分区，可以在一组消费者之间实现负责均衡。 通常当消费者被分配了一个日志分区时，它会以直接的单线程方式顺序读取分区中的消息，这种粗粒度的负载均衡有一些缺点: 1. 消费者的最大数量等于主题的分区数 2. 如果单个消息处理缓慢，会阻碍分区中后续消息的处理。可以将消息处理扩展至线程池，但这种方法消息的确认和消费者偏移管理变得复杂，通常单线程是优选。 #### 对比日志与传统消息系统 AMQP 类型的消息代理更适合如下场景: - 消息处理的代价很高，希望在逐个消息的基础上并行处理 - 而且消息排序又不那么重要的情况下，并且需要在他们处理后返回并在此读取旧消息时 更可取 相反，基于日志的消息系统更适用于: - 在消息吞吐量高，每个消息处理速度快 - 消息顺序有很重要 - 需要读取已经确认的旧消息 #### 消费者偏移量 消费者偏移量用于记录消费者已经消费的消息位置，这样就无须跟踪每条消息的确认，减少记录开销。这种偏移量与主从复制数据库中的日志序列号非常相似。 如果消费者节点失败，消费者组中的另一节点将被分配到失败的消费者分区，并以最后记录的偏移量开始使用消息。 #### 磁盘使用 不断追加的日志，磁盘空间最终会被耗尽。为了回收磁盘空间，日志实际上被分割成段，并不时地将旧段删除或归档保存。实际上，日志实现了一个有限大小的缓冲区，当缓冲区变满时，旧的消息被丢弃，该缓冲区被称为环形缓冲区。由于缓冲区在磁盘上，因此可以非常大。 #### 消费者跟不上生产者时 一开始我们讨论消费者跟不上生产者发送消息的速度时由三种选择。基于日志的方法是一种缓冲形式，它具有较大但固定大小的缓冲区。因此消费者落后太多，代理将丢弃缓冲区容量不能容纳的旧消息。可以监控消费者落后日志头部的距离，并在落后明显时发出警告。由于缓冲区很大，通常有足够的时间去修复缓慢的消费者，并允许它在开始丢失消息之前赶上。 即便落后太多并开始丢失消息也只有该消费者会受到影响，他不会中断其他消费者的服务。这是基于日志方法的消息代理带来的运营优势。在传统的消息代理中，需要小心删除消费者已经关闭的任何队列，否则他们讲继续不必要的积累消息，并占用其他活动消费者的内存。 #### 重新处理消息 使用AMQP 风格的消息代理，由于会导致消息在代理上被删除，因此处理和确认操作可视为带有一定破坏性。另一方面基于日志的消息代理中，使用消息更像是从文件读取，并不会更改日志。除了消费者的输出外，唯一的副作用是消费偏移量前移了。但是偏移量在消费者的控制之下，可以轻松对其进行操作了。 这个特点使得基于日志的消息系统更像之前讲的批处理系统过程。派生数据通过可重复的转换过程与输入数据明确分离。支持更多的实验性尝试，也更容易从错误和故障中进行恢复。从而成为集成数据流的不错选择。 ## 3. 数据库与流 事件是某个时刻发生的事情的记录，发生的事件可以是用户的操作，当然也可以是写入数据库。**将内容写入数据库的事实**是一个可以**被捕获、存储和处理的事件**。这一观察结果表明，**数据库和数据流**之间的联系比磁盘上**日志的物理存储**更紧密。 实际上**复制日志**是数据库写入事件的流，由主节点在处理事务时生成。从节点将写入流应用于他们自己的数据库副本，从而最终得到相同数据的准确副本。复制日志的事件描述了数据变化。 我们还在全序广播的中讨论了**状态机复制原理**，该原理指出: 如果每个事件代表对数据库的写入，并且每个副本按照相同的顺序处理相同的事件，则所有副本都将收敛于相同的最终状态。 ### 3.1 保持系统同步 没有一个系统能够满足所有的数据**存储、查询和处理需求**。一个大型应用通常由多个异构系统组成，这些异构系统都有自己的数据副本，以自己的表示方法存储，并针对自己的设计目标优化。由于相同和相关的数据出现在多个不同的地方，因此它们需要保持同步: 如果数据库中的某项更新，则也需要在缓存、搜索引擎、数据仓库中进行更新。 保证异构系统数据同步的一种方法是使用 ETL，即批处理。如果定期的完整数据库转储过于缓慢，有时使用的替代方法是双重写入，由程序代码在数据更改时显示的写入每个系统。但是双重写入有一些严重的问题: 1. 因为写入是有不同客户端在并发写请求同时完成的，因此可能存在竞争条件(入下图) 2. 其中一个写入可能成功，而另一个可能失败，确保他们都成功或者失败属于原子提交范畴 ![双重写入引发的竞争条件](/images/db/double_write.png) ### 3.2 变更捕获 变更数据捕获(Change Data Capture CDC) 记录了写入数据库的所有更改，兵役可复制到其他系统的形式提取数据。如果在写入时立即将更改作为一种流来发布(更改日志)，其他派生数据系统知识变更流的消费者，只要所有异构系统以相同的顺序应用于更改日志，那么所有异构系统将与数据库匹配。注意数据库系统会保证写入顺序，而在双重写入中并没有任何机制来保证多个客户端对所有系统都具有相同的写入顺序。 ![变更数据捕获](/images/db/cdc.png) #### 实现变更数据捕获 从本质上讲，变更数据捕获使得一个数据库称为主节点(捕获变化的数据库)，并将其他编程从节点。由于基于日志的消息代理保留了消息的顺序，因此它非常适合从源数据库传输更改时间。 解析复制日志一捕获变更数据的好办法。LinkedIn Databus、Facebook Wormhole 和 Yahoo Sherpa 已经大规模部署。Bottled Water 使用解码雨鞋日志的 API 实现了 PostgreSQL 的 CDC，Maxwell 和 Debezinum 通过解析 binlog 为 MySQL 做类似的事情，Mongoriver 读取 MongoDB 的 oplog，Oracle GoldenGate 也提供类似功能。 **像消息代理一样，变更数据捕获通常是异步的**: 记录系统不会再提交变更之前等待应用于消费者。这样做的优势是添加缓慢的消费者不会对记录系统产生影响，但是**所有复制滞后问题**在这里全部适用。 越来越多的数据库开始支持将变更流作为标准接口，例如 RethinkDB 支持订阅查询结果发生变化的通知，Firebase 和 CouchDB 的数据同步基于 change feed 并同时提供给应用层，而Meteor 适用 MongoDB oplog 来订阅数据更改消息并更新用户界面。 VoltDB 支持事务以流的形式连续从数据库导出数据。**kafka Connect**致力于将广泛的**数据库系统变更数据采集工具**与kafka 集成。一旦更改事件流汲取到 kafka，它就可以用来更新派生数据系统。 ## 4. 事件溯源 事件溯源类似于变更数据捕获，但是它在不同的抽象层次上应用了这个想法。事件溯源类似于编年史数据模型，其小心的区分命令和事件，用户的请求到达时，它最初是一个命令，它必须被验证。如果验证成功，它将变成一个持久且不可变的事件。 与批处理受益于其输入文件的不变形。这种不变形原则也是事件溯源和变更数据捕获如此强大的原因。 我们通常**将数据库看成是用来存储应用程序的当前状态**，这种表示**针对读取进行了优化**。每当状态变卦，改状态就反映了随着时间推移而变化的事件的结果。无论状态如何变化，总有一系列事件导致这些变化。关键思路是可变的状态和不变事件的追加日志不想矛盾。 应用状态是事件流对时间的积分得到的，而变化流式状态对时间求导得到的。 事务日志记录了对数据库所做的所有更改。高速追加式更改日志的唯一方法。从这个角度看，**数据库的内容保存了日志中最新记录值的缓存**。日志是事实，数据库是**日志子集的缓存**。日志压缩则是链接日志和数据库区别的一种方式，它保留每条记录的最新版本，并丢弃被覆盖的版本。通过不可变事件的追加日志，诊断问题和从问题中恢复就要容易得多。 ### 4.1 相同的事件日志派生多个视图 此外通过从不变事件日志中分离可变状态，可以从相同的时间日志派生出多个面向读取的表示方式。分析型数据库Druid使用这种方式直接从Kafka摄取数据，Pistachio是一个分布式的键值存储，使用Kafka作为提交日志，**Kafka Connect sinks**能将来自Kafka的数据导出到各种不同的数据库与索引。 从事件日志到数据库有一个明确的转换步骤，可以更加容易的随时间来演进应用程序: 如果想要引入一个新的方式呈现现有数据，可以使用事件日志来构建一个单独针对新功能的读取优化视图，并与现有的系统一起运行，而不要修改它们。 如果不必担心如何去查询和访问数据，那么存储数据通常是非常简单的。**模式设计、索引和存储引擎的许多复杂性**多是源于**希望支持某些查询和访问模式**。因此将数据写入形式与读取形式分开，并允许多个不同的读取视图，可以获得更大的灵活性。这个想法有时被称为**命令查询责任分离**。 数据库和模式设计的传统方法是基于**数据查询和数据写入的形式必须相同这一谬误**。如果可以将数据从针对写入优化的事件日志转换为针对读取优化的应用状态，那么有关规范化和非规范化的争论就变得无关紧要了（参阅“多对一和多对多的关系”）：由于**转换过程提供了响应机制使其与源事件日志保持一致**，因此在度优化的视图中对数据进行反规范化处理完全合理的。 ### 4.2 不可变的限制 事件捕获和变更数据捕获的最大缺点是**事件日志的消费者是异步的**，所有用户可能会写入日志，然后日志派生的视图中读取，却发现这些写操作还没有反应在读取视图中。 此外对于频换变化的数据集，保留所有变化的历史是不可行的，碎片化也可能称为一个问题，并且压缩和垃圾回收的性能对于运维的健壮性变得至关重要。 除了性能，还可能在某些情况下，由于监管方面的原因需要删除数据，尽管要求这些数据都是不可变的。比如法规要求用户关闭账户后必须删除个人所有数据。在这种情况下，标记删除是明显不够的。真正的删除数据反而会变得非常困难，因为数据副本可能在很多地方都有。 ## 5. 流处理 前面我们已经讨论了**流的来源**，讨论了**流是如何传输的**，接下来我们要讨论的是: 有了流之后，可以用它来做什么，即怎么处理它，通常有三种选择: 1. 数据同步: 将数据写入数据库，缓存，搜索引擎等其他异构系统 2. 通过某种方式推送给用户 3. 处理一个或多个输入流以产生一个或多个输出流 接下来我们将讨论选项3: 处理流以产生其他派生流。它与前面讨论 MapReduce 密切相关，并且数据流的模式是类似的: 流处理器以只读的方式接收输入流，并以仅追加方式将处理输出写入新的位置。流与批量作业的一个关键区别是，流不会结束。这种差别有很多含义: 1. 排序对无界数据集没有意义，因此不能使用排序合并 join 2. 容错机制必须改变: 对于已经运行了几分钟的批处理作业，可以简单从头开始重新执行，但是对于运行了好几年的流处理作业，重新执行几乎不可能 ### 5.1 流处理的适用场景 流长期以来一直被用于监控目的，随着时间推移也出现了如下的新用途: 1. 复杂事件处理: 2. 流分析: 按照一个时间窗口对数据进行聚合操作 3. 维护物化视图: 变更数据捕获来保持不同数据系统间的数据同步 4. 在流上搜索 5. 消息传递和 RPC ### 5.2 流的时间问题 流处理系统经常需要和时间打交道: 1. 许多流处理框架使用处理节点上的本地系统时钟(处理时间)来确定窗口，但这个有效的前提是事件发生和事件处理的间隔可以忽略不计，然而显著滞后的情况也时有发生。 #### 事件时间和处理时间 事件处理时间比发生时间滞后的原因有很多，例如排队、网络故障、重启动消费者、重新处理过去的事件等等。而且消息延迟还可能导致**消息的不可预知排序**。**混淆事件发生时间和处理会导致错误的结果**。 #### 什么时候准备就绪 如果基于事件发生时间而定义窗口，面临一个棘手的问题是，你无法确定**什么时候能收到特定窗口内的所有事件**，或者**是否还有一些事件尚未到来**。由于网络中断和延迟，可能某些发生的事件还缓存在另一台机器上。需要能够处理**在窗口已经声明完成后才到达的这样的滞后事件**。大体上行有两种选择: 1. 忽略这些滞后事件 2. 发布一个更正: 针对滞后事件的一个更新值 #### 你用谁的时间 当事件可能在系统的多个点缓冲时，为事件分配事件比较困难。考虑一个可离线使用的应用: 1. 事件可能在本地离线缓存，在下一次连接网络时被发送至服务器。这时事件的时间戳实际上指的是发生交互时的时间，然后用户设备上的时间通常是不可信的 2. 服务器收到事件的时间可能更准确，但在描述用户交互方面的意义不大 为了调整不正确的设备时钟，一种方法是记录三个时间戳: 1. 根据设备的时钟，记录事件发生的时间 2. 根据设备的时钟，记录事件发送到服务器的时间 3. 根据服务器时钟，记录服务器收到事件的时间 3-2 可以估计出设备和服务器时钟之间的偏移量，从而估算出事件实际发生的时间。 ### 5.3 窗口类型 一旦明确了如何确定事件的时间戳，下一步就是决定如何定义时间段即窗口，有如下几种常见的窗口类型: 1. 轮转窗口: 翻滚窗口的长度固定，每一个事件都属于一个窗口 2. 跳跃窗口: 跳跃窗口也有固定长度，但允许窗口重叠以提供一些平滑过渡 3. 滑动窗口： 滑动窗口包含在彼此的某个时间间隔内的所有事件 4. 会话窗口：没有固定的持续时间，通过将同一用户在时间上机密相关的所有事件分组在一起而定义 ### 5.3 流式 join 因为流的无界，以及新事件随时可能出现在流中，使得流式join比批处理作业更具挑战性，我们区分了三种不同类型的join: 1. 流与流join 2. 流与表join 3. 表与表join #### 流与流join #### 流与表join #### 表与表join 这三种类型的 join 有很多相似之处，它们都需要流处理系统根据一个 join 输入 来维护某些状态，并在来自另一个 join 输入的消息上查询状态。 维持这个状态的事件的顺序非常关键。在分区日志中单个分区内的事件顺序被保留，但是通常不同流或分区之间没有排序保证。这就产生一个问题: **如果不同流中的事件发生在相近的时间里，它们按照何种顺序进行处理**。在流与表的join 中，如果用户更新了资料，哪些活动事件会与旧资料join，哪些又与新资料join。换句话说，如果状态随时间改变，join 操作需要输入状态信息，那么应该使用什么时间点的状态来 join 呢？ 如果垮流的时间排序是不确定的，那么 join 也变成非确定性的。这意味着**不能在相同的输入上重新运行同一作业来得到相同的结果**: 因为再次运行任务时，输入流上的事件可能以不同的方式交叉在一起。 ### 5.4 流处理容错 流处理面临和批处理一样失败任务重试的问题。但是处理起来并不简单: 流处理无法像批处理一样，在使输出结果可见之前等待某个任务完成，因为流是无限的，因此几乎永远无法完成这个任务。 #### 微批处理和校验点 一种解决方案是将流分解成多个小块，冰箱小型批处理一样处理每个块。这种方法称为微批处理，已经用于 SparkStreaming。批处理大小通常约为 1s，这是一个折中的考虑: 1. 较小的批处理会导致更大的调度和协调开销 2. 较大的批处理意味着流处理器的结果需要更长的延迟才能看见 微批处理隐含的设置了与批处理大小相等的轮转窗口(基于处理时间而不是事件时间)。任何需要更大窗口的作业都需要显示的将状态从一个微批处理保留至下一个微批处理。 Apache Flink 中使用了该方法的一个变体，它定期生成状态滚动检查点并将其写入持久化存储。如果流操作发生崩溃，他可以从最近的检查点重新启动，并丢弃检查点和崩溃之间的所有输出。检查点由消息流中的barrier 触发，类似于微批处理的边界，但不强制特定的窗口大小。 在流处理框架内，微批处理和检查点提供了与批处理一样的恰好一次语义，但是一旦输出脱离了流处理系统，框架将无法丢失失败批处理的输出。 微批处理存在处理之后消息和跨批量边界的窗口等问题。 #### 重新审视原子提交 在出现故障时，为了看起来实现**恰好一次**语义，我们需要确保当且仅当处理成功时，所有输出和副作用才生效。这包括发送给下游的操作和外部消息传递系统的任何消息。所有数据库写入，以及对操作状态的更改和任何对输入消息的确认。 这些事要么原子的发生，要么都不发生，我们在之前分布式事务和共识中已经有所讨论。 #### 幂等性 我们的目标是**丢弃任何失败的部分输出**，以便可以安全的重试而不会两次生效。分布式事务是实现的一种方式，另一种方式则是依赖幂等性。即使操作本身不具有幂等，往往也可以使用一些额外的元数据(唯一键)使其变得幂等。 依赖幂等性意味着基于一些假设: 重启失败的任务必须以相同的顺序重新处理相同的消息，处理必须是确定的，**并且没有其他节点并发更新相同的值**。 **任何需要状态的流处理，都必须确保在故障发生后状态可以恢复**。有两种选择: 1. 将状态保存在远程存储中并采取复制，然而为每个消息去查询远程数据库可能会很慢 2. 将状态保存在本地，并定期进行复制 当流处理器从故障中恢复时，新任务可以读取副本并在不丢失数据的情况下恢复处理。在某些情况下，甚至可能不需要复制状态，而是从输入流开始重建。 所有权衡取决于底层基础架构的性能表现，网络延迟和磁盘访问延迟，网络带宽和磁盘带宽到底谁大，都是需要权衡的因素。