CAP 定理是数据系统设计的基本理论

这个函数产生的是 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 定理交互得也很糟糕）。