Twitter 首席工程师：如何“打败”CAP 定理

xman

CAP 定理是数据系统设计的基本理论，目前几乎所有的数据系统的设计都遵循了这个定理。但 CAP 定理给目前的数据系统带来了许多复杂的、不可控的问题，使得数据系统的设计越来越复杂。Twitter 首席工程师、Storm 的作者 Nathan Marz 在本文中通过避开 CAP 定理带来的诸多复杂问题，展示了一个不同于以往的数据系统设计方案，给我们的数据系统设计带来了全新的思路。


　　CAP 定理指出，一个数据库不可能同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition-Tolerance）。


　　一致性（Consistency）是指执行了一次成功的写操作之后，未来的读操作一定可以读到这个写入的值。可用性（Availability）是指系统总是可读可写的。Yammer 的 Coda Hale 和 Cloudera 的 Henry Robinson 都阐述过，分区容错性是不能牺牲的，因此只能在一致性和可用性上做取舍，如何处理这种取舍正是目前 NoSQL 数据库的核心焦点。


　　选择一致性而不是可用性的系统将面临一些尴尬的问题，当系统不可用时怎么办？你可以对写操作进行缓冲处理，但如果存储缓冲数据的机器出现故障，客户端将丢失写入的值。同样地，缓冲写也可以被认为是一种非一致性的操作，因为客户端认为成功的写入实际上并没有写入到实际的数据库中。当然，系统可以在机器不可用时向客户端返回错误，但可以想象，一个经常告诉客户端“请重试”的产品是多么令人讨厌。


　　另一个方案是选择可用性放弃一致性。这种情况下最好的一致性保障是“最终一致性”（Eventually Consistency）。当使用最终一致性的系统时，客户端有时会读到与刚刚写入数据不同的数据。有时候，同一时间同一个 key 的多个请求有可能返回不同的结果。数据更新并不能及时在所有的复制节点上生效，所以不同的复制节点上可能读取到的是不同的值。当你检测到数据不一致性时，你需要进行修复（Repair）操作，这就需要使用矢量时钟（vector clock）记录数据的版本历史并合并不同的数据更新（这称为读取修复，read repair）。


　　我相信在应用层维护最终一致性对开发人员负担太重，开发人员极易弄错读取修复的代码，而一旦开发人员犯错，有问题的读取修复将对数据库系统造成不可逆的损坏。


　　所以牺牲可用性时问题会很多，牺牲一致性时构建和维护系统的复杂度又很高，但这里又只有两个选择，不管怎样做都会不完美。CAP 定理是改不了的，那么还有什么其他可能的选择吗？


　　实际上，还有一个办法：你并不能避开 CAP 定理，但可以把复杂的问题独立出来，免得你丧失对整个系统的掌控能力。CAP 定理带来的复杂性，其实是我们如何构建数据系统这一根本问题的体现。其中有两点特别重要：数据库中可变状态和更新状态的增量算法。复杂性正是这两点和 CAP 定理之间的相互作用导致的。


　　本文将通过一个数据库系统的设计，来说明如何解决 CAP 定理通常会造成的复杂性问题。但我要做的不仅仅如此，CAP 定理是一个针对机器发生错误时系统容错性的一个定理，而这里有比机器容错性更加重要的容错性——人为操作容错性。在软件开发中一个确定的事实是，开发人员都并非完人，产品中难免有一些 Bug，我们的系统必须对有 Bug 的程序写入的错误数据有足够的适应能力，我要展示的系统将是这样一个可以容忍人为错误的系统。


　　本文将挑战你对数据系统如何构建这一问题的假设，通过颠覆传统数据系统构建方法，我会让大家看到一个前所未见的优雅、扩展性强、健壮的数据系统。


　　什么是数据系统？


　　在开始介绍系统设计之前，让我们先来看看我们要解决的问题：数据系统的目的在于什么？ 什么是数据？ 在我们考虑 CAP 定理之前，我们必须给出一个可以适用于所有数据应用程序的定义来回答上述问题。


　　数据应用程序种类很多，包括存入和提取数据对象、连接、聚合、流处理、机器学习等。似乎并不存在一个对数据系统的明确定义，数据处理的多样性使得我们很难用一个定义来描述。


　　事实却并非如此，下面这个简单的定义：


　　Query = Function (All Data)


　　概括了数据库和数据系统的所有领域。每一个领域——有 50 年历史的 RDBMS、索引、OLAP、OLTP、MapReduce、EFL、分布式文件系统、流处理器、NoSQL 等——都可以被概括进这个方程。


　　所谓数据系统就是要回答数据集问题的系统，这些问题我们称之为“查询”。上面的方程表明，查询就是数据上的一个函数。


　　上述方程对于实际使用来说太过于笼统，几乎对复杂的数据系统设计不起什么作用。但如果所有的数据系统都遵循这个方程又会怎样呢？这个方程是探索我们数据系统的第一步，而它最终将引导我们找到“打败”CAP 定理的方法。


　　这个方程里面有两个关键概念：数据、查询。这两个完全不同的概念经常被混为一谈，所以下面来看看这两个概念究竟是什么意思。


　　数据


　　我们先从“数据”开始。所谓数据就是一个不可分割的单位，它肯定存在，就跟数学里面的公理一样。


　　关于“数据”有两个关键的性质。首先，数据是跟时间相关的，一个真实的数据一定是在某个时间点存在于那儿。比如，假如 Sally 在她的社交网络个人资料中写她住在芝加哥，你拿到的这个数据肯定是她某个时间在芝加哥填写的。假如某天 Sally 把她资料里面居住地点更新为亚特兰大，那么她肯定在这个时候是住在亚特兰大的，但她住在亚特兰大的事实无法改变她曾经住在芝加哥这个事实——这两个数据都是真实的。


　　其次，数据无法改变。由于数据跟某个时间点相关，所以数据的真实性是无法改变的。没有人可以回到那个时间去改变数据的真实性，这说明了对数据操作只有两种：读取已存在的数据和添加更多的新数据。那么 CRUD 就变成了 CR【译者注：CRUD 是指 Create Read Update Delete，即数据的创建、读取、更新和删除】。


　　我去掉了“更新”操作，因为更新对于不可改变的数据没有任何作用。例如，更新 Sally 的位置信息本质上就是在她住的地方数据中新加一条最近的位置信息而已。


　　我同样去掉了“删除”操作，因为绝大部分删除操作可以更好地表述为新加一条数据。比如 Bob 在 Twitter 上不再关注 Mary 了，这并不能改变他曾经关注过 Mary 这个事实。所以与其删除 Bob 关注 Mary 这个数据，还不如新加一条 Bob 在某个时间点不再关注 Mary 这个数据。


　　这里只有很少数的情况需要永久“删除”数据，例如规则要求你每隔一段时间清掉数据，这个情况在我将要展示的系统中有很好的解决方案，所以为了简洁，我们暂不考虑这些情况。


　　查询


　　查询是一个针对数据集的推导，就像是一个数学里面的定理。例如，你可以通过计算“Sally 现在的位置在哪里”这个查询来得到 Sally 最新的位置数据。查询是整个数据集合上的函数，可以做一切事情：聚合、连接不同类型的数据等。因此，你可以查询系统中女性用户的数量，可以查询最近几小时热门的 Twitter 内容。


　　前面我已经定义查询是整个数据集上的函数，当然，不是所有的查询都需要整个数据集，它们只需要数据集的一个子集。但我的定义是涵盖了所有的查询类型，如果想要“打败”CAP 定理，我们需要能够处理所有的查询。


　　打败 CAP 定理


　　计算查询最简单的办法就是按照查询语义在整个数据集上运行一个函数。如果这可以满足你对延迟的要求，那么就没有其他需要构建的了。


　　可想而知，我们不能指望在整个数据集上的查询能够很快完成，特别是那些服务大型网站、需要每秒处理几百万次请求的系统。但假如这种查询可以很快完成，让我们来看看像这样的系统和 CAP 定理的 PK 结果：你将会看到，这个系统不仅打败了 CAP 定理，而且还消灭了它。


　　CAP 定理仍然适用，所以你需要在可用性和一致性上做出选择，这里的漂亮之处在于，一旦你权衡之后做出了选择，你就做完了所有的事情。通常的那些因为 CAP 定理带来的问题，都可以通过不可改变的数据和从原始数据中计算查询来规避。


　　如果你选择一致性而不是可用性，那么跟以前并没有多大的区别，因为你放弃了可用性，所以一些时候你将无法读取或者写入数据。当然这只是针对对强一致性有要求的系统。


　　如果你选择可用性而不是一致性，在这种情况下，系统可以达到最终一致性而且规避了所有最终一致性带来的复杂问题。由于系统总是可用的，所以你总可以写入新数据或者进行查询。在出错情况下，查询可能返回的不是最近写入的数据，但根据最终一致性，这个数据最终会一致，而查询函数最终会把这个数据计算进去。


　　这里的关键在于数据是不可变的。不可变数据意味着这里没有更新操作，所以不可能出现数据复制不同这种不一致的情况，也意味着不需要版本化的数据、矢量时钟或者读取修复。在一个查询场景中，一个数据只有存在或者不存在两种情况。这里只有数据和在数据之上的函数。这里没有需要你为确保最终一致性额外做的事情，最终一致性也不会因此使你的系统变得复杂。


　　之前的复杂度主要来自增量更新操作和 CAP 定理之间的矛盾，在最终一致性系统中可变的值需要通过读取修复来保证最终一致性。通过使用不可变数据，去掉增量更新，使用不可变数据，每次从原始数据计算查询，你可以规避那些复杂的问题。CAP 定理就被打败了。


　　当然，现在讲的只不过是想法而已，而且每次从原始数据计算查询基本上不可能。但我们从中可以学到一些在实际解决方案中的关键点。


数据系统因为不可变数据和不断增长的数据集变得简单了。
基本的写入操作就是写入一条新的不可变数据。
数据系统通过重新从原始数据计算查询规避了 CAP 定理带来的复杂度。
数据系统利用增量算法使得查询的返回延迟降低到一个可以接受的程度。

　　让我们开始探索这个数据系统应该如何设计。请注意从这里开始我们所描述都是针对系统优化、数据库、索引、EFL、批量计算、流处理——这些技术都是对查询函数的优化，让查询返回时间降低到一个可以接受的程度。这很简单，但也是数据系统所面对的现实。数据库通常是数据管理的核心，但它们是更大蓝图中的一部分。


　　批量计算


　　“如何让任意一个函数可以在任意一个数据集上快速执行完成”这个问题太过于复杂，所以我们先放宽了一下这个问题依赖条件。首先假设，可以允许数据滞后几小时。放宽这个条件之后，我们可以得到一个简单、优雅、通用的数据系统构建解决方案。之后，我们会通过扩展这个解决方案使得它可以不用放宽条件来解决问题。


　　由于查询是所有数据的一个函数，让查询变快的最简单的方法就是预先计算好这些查询。只要这里有新的数据，你就重新计算这些查询。这是可能的，因为我们放宽了条件使得我们的数据可以滞后几个小时。图 1 展示了这个工作流程。



　图 1 预计算工作流程
　　为了实现这个，你的系统需要：


能很容易存储大的、不断增长的数据集；
能在数据集上可扩展地计算查询函数。

　　这样的系统是存在的，即 Hadoop。它是一个成熟的、经历了无数团队实战检验过的系统，同时拥有一个巨大的工具生态系统。它虽不完美，但是这里用来做批量处理的最好的一个工具。


　　许多人也许会告诉你，Hadoop 只适用于那些“非结构化”的数据，这是完全错误的看法。Hadoop 处理“结构化”的数据也很不错，通过使用像 Thrift 或者 Protocol Buffers 这样的工具，你可以使用丰富的数据结构存储你的数据。


　　Hadoop 由分布式文件系统 HDFS 和批处理框架 MapReduce 两部分构成。HDFS 可以通过文件存储大量数据，MapReduce 可以在这样数据上进行可扩展计算。这个系统完全符合我们的要求。


　　我们将数据以文件形式存储到 HDFS 中去。文件可以包括一个数据记录序列。新增数据时，我们只需要在包括所有数据的文件夹中新增一个包含这条新记录的文件即可。像这样在 HDFS 存储数据满足了“能够很容易存储大的、不断增长的数据集”这个要求。


　　预计算数据集上的查询也很直观，MapReduce 是一个足够复杂的框架，使得几乎所有的函数都可以按照多个 MapReduce 任务这种方式实现。像 Cascalog、Cascading 和 Pig 这样的工具使实现这些函数变得十分简单。


　　最后，为了可以快速访问这些预计算查询结果，你需要对查询结果进行索引，这里有许多数据库可以完成这个工作。ElephantDB 和 Voldemort read-only 可以通过从 Hadoop 中导出 key/value 数据来加快查询速度。这些数据库支持批量写和随机读，同时不支持随机写。随机写使得数据库变得复杂，所以通过不支持随机写，这些数据库设计得特别简洁，也就几千行代码而已。简洁使得这些数据库鲁棒性变得非常好。


　　下面来看批量处理系统整体上是如何配合工作的。假设写一个网站分析程序来跟踪页面访问量，你需要能够查询到任意时间段的页面访问量，数据是以小时方式提供的。如图 2 所示。



　　图 2 批处理工程流程示例(timestamp 代表时间戳，count 代表个数)


　　实现这个很简单，每一个数据记录包括一个单一页面的访问量。这些数据通过文件形式存储到 HDFS 中，一个函数通过实现 MapReduce 计算任务，来计算一个 URL 下页面每小时的访问量。这个函数产生的是 key/value 对，其中[URL, hour]是 key，value 是页面的访问量。这些 key/value 对被导出到 ElephantDB 中去，使得应用程序可以快速得到任意[URL, hour]对对应的值。如果应用程序想要知道某个时间范围内某个页面的访问量，它可以查询 ElephantDB 中那段时间内的数据，然后把这些数据相加就可以得到这个访问量数据了。


　　在数据滞后几小时这个缺陷下，批量处理可以计算任意数据集上的任意函数。系统中的“任意性”是指这个系统可以处理任何问题。更重要的是，它很简单，容易理解和完全可扩展，你需要考虑的只是数据和查询函数，Hadoop 会帮你处理并行的事情。


　　批处理系统、CAP 定理和容忍人为错误


　　截至目前，我们的系统都很不错，这个批处理系统是不是可以达到容忍人为错误的目标呢？


　　让我们从 CAP 定理开始。这个批处理系统总是最终一致的：写入的数据总可以在几小时后被查询到。这个系统是一个很容易掌控的最终一致性系统，使得你可以只用关注你的数据和针对数据的查询函数。这里没有涉及读取修复、并发和其他一些需要考虑的复杂问题。


　　接下来看看这个系统对人为错误的容忍性。在这个系统中人们可能会犯两个错误：部署了一个有 Bug 的查询函数或者写入了错误的数据。


　　如果部署了一个有 Bug 的查询函数，需要做的所有事情就是修正那个 Bug，重新部署这个查询函数，然后在主数据集上重新计算它。这之所以能起作用是因为查询只是一个函数而已。


　　另外，错误的数据有明确的办法可以恢复：删除错误数据，然后重新计算查询。由于数据是不可变的，而且数据集只是往后添加新数据，写入错误的数据不会覆盖或者删除正确的数据，这与传统数据库更新一个数据就丢掉旧的数据形成了鲜明的对比。


　　注意到 MVCC 和 HBase 类似的行版本管理并不能达到上面人为错误容忍级别。MVCC 和 HBase 行版本管理不能永久保存数据，一旦数据库合并了这些版本，旧的数据就会丢失。只有不可变数据系统能够保证你在写入错误数据时可以找到一个恢复数据的方法。


　　实时层


　　上面的批量处理系统几乎完全解决了在任意数据集上运行任意函数的实时性需求。任何超过几个小时的数据已经被计算进入了批处理视图中，所以剩下来要做的就是处理最近几个小时的数据。我们知道在最近几小时数据上进行查询比在整个数据集上查询要容易，这是关键点。


　　为了处理最近几个小时的数据，需要一个实时系统和批处理系统同时运行。这个实时系统在最近几个小时数据上预计算查询函数。要计算一个查询函数，需要查询批处理视图和实时视图，并把它们合并起来以得到最终的数据。




图 3 计算一个查询
　　在实时层，可以使用 Riak 或者 Cassandra 这种读写数据库，而且实时层依赖那些数据库中对状态更新的增量算法。


　　让 Hadoop 模拟实时计算的工具是 Storm。我写 Storm 的目的是让 Hadoop 可以健壮、可扩展地处理大量的实时数据。Storm 在数据流上运行无限的计算，并且对这些数据处理提供了强有力的保障。


　　让我们回到刚才那个根据某个 URL 查询某个页面在某个时间段内页面访问量的例子，通过这个例子我将展示实时层是如何工作的。


　　批处理系统还是跟之前一样：一个基于 Hadoop 和 ElephantDB 的批处理工作流，在几个小时之前的数据上预计算查询函数。剩下就是让实时系统去处理最近几小时数据了。


　　我们将最近几小时的数据状态存入 Cassandra 中，用 Storm 去处理页面访问量数据流并并行更新到数据库中，针对每一个页面访问量，在[URL, hour]所代表的 key 下，有一个计数器，这个计数器在 Cassandra 中实现。这就是所有的事情，Storm 让事情变得非常简单。



　图 4 批处理/实时架构示例
　　批处理层+实时层、CAP 定理和人为错误容忍性


　　貌似又回到一开始提出的问题上去了，访问实时数据需要使用 NoSQL 数据库和增量算法。这就说明回到了版本化数据、矢量时钟和读取修复这些复杂问题中来。但这是有本质区别的。由于实时层只处理最近几小时的数据，所有实时层的计算都会被最终批处理层重新计算。所以如果犯了什么错误或者实时层出了问题，最终都会被批处理层更正过来，所有复杂的问题都是暂时的。


　　这并不意味着不需要关心实时层的读取修复和最终一致性，你仍然需要实时层尽可能的一致。但当犯了一个错误时，不会永久性地破坏数据。这便移除了许多你所需要面对的复杂问题。


　　在批处理层仅需要考虑数据和数据上的查询函数，批处理层因此很好掌控。在实时层，需要使用增量算法和复杂的 NoSQL 数据库。把所有的复杂问题独立到实时层中，对系统的鲁棒性、可靠性做出了重大贡献。


　　同样的，实时层并没有影响系统的人为错误容忍性，这个数据不可变和只追加的批处理系统，仍然是整个系统的核心，所以所有的都可以像上面说的一样被纠正过来。


　　我有一个类似的系统：Hadoop 和 ElephantDB 组成批处理系统，Storm 和 Cassandra 组成实时系统。由于缺乏监控，某天当我起床的时候发现 Cassandra 运行满负荷了，使得所有的数据请求都超时。这使得 Storm 计算失败，一些数据又重新回到了等待队列中，这个数据就一次次重复请求。


　　如果我没有批处理层，那么我就需要扩展和恢复 Cassandra，这个很不容易。更糟的是，因为请求不断的重复，无法得到正确的数据。


　　幸运的是，所有的复杂问题都被隔离到实时层中去了，我清空了所有的后台请求队列，把它们打到了批处理层上，同时重启了 Cassandra 集群，过了几个小时之后所有数据都恢复正常了。没有错误数据，请求中也没有不准确的地方。


　　垃圾回收


　　上面描述的所有东西都是建立在一个不可变的、不断增长的数据集上的。如果数据集已经很大，使得不可能用水平扩展储存所有时间的所有数据，该如何处理呢？这是不是就推翻了我说的一切呢？是不是需要回到可变数据的系统上呢？


　　不。我们可以很容易地用“垃圾回收”对基本模型进行扩展来解决上面的问题。垃圾回收是一个在主数据集上的简单函数，返回的是一个过滤版本的主数据集。垃圾回收掉了旧数据，可以选择任意的垃圾回收策略。可以在易变的系统中只保留数据最新的一个值或者保留每个数据的历史。比如，如果要处理位置数据，可以保留每人每年的一个地点。可变性是一个不是很灵活的垃圾回收形式（它跟 CAP 定理交互得也很糟糕）。


　　垃圾回收可以被实现成批处理的一个任务，隔段时间运行一下。由于它是作为离线批处理任务执行的，所以不影响我们与 CAP 定理的交互。


　　总结


　　让可扩展的数据系统复杂的原因不是 CAP 系统，而是数据增量算法和数据的可变状态。最近由于分布式数据库的兴起导致了复杂度越来越不可控。前面讲过，我将挑战对传统数据系统构建方法的假设。我把 CRUD 变成了 CR，把持久化层分成了批处理和实时两个层，并且得到对人为错误容忍的能力。我花费了多年来之不易的经验打破我对传统数据库的假设，并得到了这些结论。


　　批处理/实时架构有许多有趣的能力我并没有提到，下面我总结了一些。


　　算法的灵活性。随着数据量的增长，一些算法会越来越难计算。比如计算标识符的数量，当标识符集合越来越大时，将会越来越难计算。批处理/实时分离系统给了你在批处理系统上使用精确算法和在实时系统上使用近似算法的灵活性。批处理系统计算结果会最终覆盖实时系统的计算结果，所以最终近似值会被修正，而你的系统拥有了“最终精确性”。


　　数据结构迁移变得很容易。数据结构迁移的难题将一去不复返。由于批量计算是系统的核心，很容易在整个系统上运行一个函数，所以很容易更改你数据的结构或者视图。


　　简单的 Ad-Hoc 网络。由于批处理系统的任意性，使得你可以在数据上进行任意查询。由于所有的数据在一个点上都可以获取，所以 Ad-Hoc 网络变得简单而且方便。


　　自我检查。由于数据是不可变的，数据集就可以自我检查。数据集记录了它的数据历史，对于人为错误容忍性和数据分析很有用。


　　我并没有说我已经“解决”了数据量过大的问题，但我已经为解决大数据问题制订了一个框架。批处理/实时架构可以应用到任何一个数据系统中去，“授人以鱼，不如授人以渔”，我已经告诉你了如何去构建这样的系统。


为了提高系统整体能力来解决大数据的问题，我们还有许多工作需要做。
扩展数据模型，支持批量写和随机读。不是每一个应用程序都支持 key/value 的数据库，这也是我们团队对扩展 ElephantDB，使得可以支持搜索、文档数据库、区间查询感兴趣的原因。
更好的批处理原语。Hadoop 并不是批处理的最终形态，好多批处理计算 Hadoop 效率不高。Spark 是一个有意思的扩展 MapReduce 的项目。
提升后的读写 NoSQL 数据库。这里不同类型数据的数据库还有很大的提升空间，随着这些数据库的成熟，它们将收获很多。
高层级的抽象。未来工作中最有意思的就是对批处理模块和实时处理模块的高层次抽象，在批处理和实时架构下你没有理由不拥有一个简单的、描述性的、鲁棒性好的语言。

　　许多人需要一个可扩展的关系型数据库，本文就是想让你知道完全不需要那个。大数据量和 NoSQL 运动使数据管理比 RDBMS 更加复杂。那仅仅是因为我对大数据的处理采用了跟 RDBMS 同样的方法：把数据和视图混为一谈，并且依赖增量算法。大数据量需要采用完全不同的方式构建数据系统。通过存储持续增长的不可变数据，并且系统核心采用预计算，大数据系统就可以变得比关系型数据库更易掌控，并且可扩展性很强。