Avinash Lakshman , Facebook Prashant Malik，Facebook
张鹏@Sina RDC 译
摘要 ABSTRACT
Cassandra 是一个分布式的存储引擎，用来管理分布在大量普通商用级别服务器上面的海量的结构化数据，可以提供高可用性，不存在单点故障。Cassandra设计目标，是运行在千台规模的服务器节点上面，节点可以跨越IDC.在这个规模上，大小组件都会频繁的发生故障。当故障发生时，Cassandra通过对持久层状态的有效管理，来达成整个系统的可靠性和扩展性。在很多场合，Cassandra作为一个数据库来使用，因此他借鉴了很多数据库的设计和实现策略，但是他不能支持完整的关系数据库模型；相反，他提供给客户端一个简单的数据模型，客户端可以通过这个模型，来动态控制数据的布局和格式。Cassandra系统设计成可以运行在大量廉价商用机上面，具备高写吞吐量的同时，不牺牲读性能。
1.介绍 INCRODUCTION
Facebook是最大的社交网络平台，在峰值的时候，他可以通过部署在世界各地很多数据中心的几万台服务器为几亿用户提供服务。Facebook的平台，为满足系统性能，可靠性和效率，为满足业务上的持续增长的扩展性，对运维提出了严格的需求。处理具有千台的节点规模的基础架构的故障，已经成为我们工作的常态。另外还有很多小的节点和网络组件，在任何时候也都会发生故障。因此软件系统在设计的时候，要把这些节点故障当成常态，而不是例外来处理。 为了应对上述可靠性和扩展性挑战，Facebook开发了Cassandra.
Cassandra采用了一系列众所周知的技术，来达到扩展性和可用性。 Cassandra 被设计成收件箱搜索的存储部分。用户通过收件箱搜索功能，来完成日常收件箱搜索操作。在Facebook,这意味着系统要能够应对非常高的写吞吐量，每天会有数十亿的写请求，这个数字还在随着用户的增长而不停增长。因为Facebook的数据中心，分布在不同的地域为用户提供服务，因此在IDC之间复制数据，是降低搜索延迟的关键。收件箱搜索在2008年6月上线，当时有1亿用户，到今天(论文发表时间)，Facebook有2.5亿用户,Cassandra仍旧能够满足需求。 Cassandra目前为Facebook的多个服务提供后端存储支持。
这个论文按照如下结构组织. 第二章描述了相关的工作，都是在我们的设计中非常重要的方面。第三章详细阐述了数据结构。第四章简要介绍了客户端API。 第五章披露了分布算法和系统设计细节。第六章详细介绍了如何搭建Cassandra系统和系统性能调优。 第六章第一节介绍了Facebook平台如何使用Cassandra 。最后第七章总结了Cassandra的后续工作。
2.相关工作 RELATED WORK
文件系统和数据库社区，对于通过分布数据方式来实现性能，可用性，可靠性，进行了广泛的研究。不同于P2P存储系统只能支持扁平的命名空间，分布式文件系统支持层级式的命名空间。像Ficus和Coda，通过复制文件来达到高可用性，但是与此同时，系统牺牲了一致性。对于更新冲突，一般通过专门的冲突解决程序来处理。Farsite 在未采用任何中心化服务器的同时，实现了分布式文件系统。Farsite通过复制技术，来实现高可用性和可扩展性。 GFS是另外一个分布式存储系统，用来存储Google内部程序数据。 GFS采用了一个简单的设计，通过一个Master服务器来存储全部的metadata. 客户的数据被切分成数据块，存储在块存储服务器上。当然现在Google通过Chubby实现了Master服务器的容灾。Bayou 是一个分布式的关系型数据库，允许节点离线，提供实现最终一致性保证。 在上面这些系统中，Bayou,Coda,Ficus 允许节点离线操作，系统能够弹性的处理网络分区和运行中断。 这些系统在冲突处理的方式上存在差别。比如，Coda 和 Ficus实现系统层面的冲突解决。Bayou允许应用程序层面进行冲突解决，所有这些系统，都能够实现最终一致性。类似这些系统, Dynamo 在发生网络分区的时候，仍旧允许读写操作，然后通过不同的冲突处理机制解决冲突，有一些机制，是客户端驱动的。传统的关系型数据库的复制技术，关注于确保复制数据的强一致性。虽然强一致性对于应用编写者来说，是个方便的编程模型，但是这些系统因为强一致性的保证，而不能应付网络分区的情况。
Dynamo是Amazon用来存取购物车的存储系统。 Dynamo 基于成员算法的 GOSSIP协议，帮助系统中每个节点维护其他全部节点的信息。 可以把Dynamo理解成一个结构化的层，请求最多通过1跳路由达到目的。Dynamo通过时钟向量图来检测更新冲突，优先采用客户端来解决冲突的机制。Dynamo中的写操作执行前，需要一个读操作来获取时间戳向量，这个特点，在系统需要高的写吞吐量的时候，会成为瓶颈。 Bigtable提供结构化和数据的分布式，但需要依赖于一个分布式文件系统来实现持续服务。
3.数据模型 DATA MODEL
Cassandra中的表，是一个分布式的多维MAP, 通过一个key进行索引。值是一个高度结构化的对象。 表中每一行的key,是一个没有大小限制的字符串，一般是16-32字节长度。在每个副本中通过Key对每一行的操作，不管进行多少列得读写都能保证原子操作。列，会被分组到SET里面，分组以后得列，被称为 列族 , 就像BigTable一样。Cassandra公开了两种列族类型，简单列族和超级列族。超级列族可以想象成，列族的嵌套结构。而且，应用程序还可以指定超级列族，列族里面的列的排序顺序，排序顺序可以按照时间，名字。通过时间对列来排序，是为了满足收件箱搜索这类应用开发出来的。通过Column_family :column形式，来访问列族里面的列，可以通过Column_family :super_column : column 来访问超级列族里面的列。
我们在Section6.1 会用一个很好的例子，来展示超级列族的抽象能力。 通常，应用会使用专有的Cassandra集群作为他们服务的一部分。 虽然系统支持多表的概念，但是目前所有的部署还都是单表部署。
4.API
Cassandra API包含下面三个简单方法。
.. insert(table , key , rowMutation)
columnName 可以指向列族中的一个列，一个列族，一个超级列族或者 超级列族中的一列。
5.系统架构 SYSTEM ARCHITECTURE
在生产环境中运行的存储系统的架构是非常复杂的。除了现行的数据存储组件以外，系统还需要满足如下要求。具有足够健壮性和扩展性来支持负载均衡，节点间关系维护和故障检测，故障恢复，同步复制，过载处理，状态传输，并发和任务调度，请求封装，请求路由，系统监控，配置管理。详细的介绍这些解决方案超出了本文的范围，因此我们在本文介绍Cassandra中分布式存储的核心技术: 分区，复制，节点关系，失效处理和扩容。这些模块协同工作，处理读写请求。一般来讲，对于一个key的读写请求，会路由到Cassandra集群的某个具体节点上面。这个节点，能够决定请求的副本节点。对于写来说，系统将请求路由到副本上面，等待最少的副本节点【编辑:最少的副本节点，既能维持系统一致性的最低的副本的数量】完成写请求。对于读请求，根据客户端对于一致性的要求，系统或者将请求路由到最近的副本节点，或者路由到所有的节点，等待有效的节点返回结果。
5.1分区 Partitioning
Cassandra的一个关键特性，是可以规模扩容。这就要求，系统能够动态在节点之间分割数据。Cassandra通过一致性的有序哈希算法，来分割数据。一致性的哈希函数中，输出的值域，在一个固定的环形空间中（哈希的最大值 紧邻着哈希的最小值）。在系统中，每个节点都会被随机分配一个值，用来标定它在环中的位置。通过哈希数据的key，来定位数据所在节点的位置。然后按照顺时针的循序，从数据节点的位置开始，找到第一个编号大于数据节点编号的节点。这个节点就是这个key的调度节点。应用程序指定这个key，然后Cassandra通过这个key来路由请求。因此，每个节点都对 环中他和他的前任节点之间的区域负责。一致性哈希规则的好处就是，一旦有新的节点加入，或者有节点离线退出，那么受影响的就是节点相邻的节点，其他的节点不受影响。基本的一致性哈希算法存在一些问题。第一，随机的分配节点的位置，会导致数据和节点负载的不均衡。第二，基本算法没有考虑到节点之间的性能差异。目前有两种方案解决这些问题，第一，像dynamo一样，为每个节点在环中分配多个位置。第二，分析环的负载情况，将负载较轻的节点，移动到负载较重的节点附近。Cassandra采用第二种方案，这种方案设计和实施上，都有非常好的可追踪性，另外在做负载均衡时，可以提供非常有效的决策数据。
5.2复制 Replication
Cassandra通过复制技术，来实现高可用性和持续服务能力。 每个数据项目都会在N个机器上做复制, N被称为复制因子，通过参数per-instance来配置。每个key都会赋值给调度节点k，调度节点负责在他的控制范围内的节点的数据复制工作。除了在调度节点控制范围之内复制数据项目以外，调度节点还会在环中N-1节点做数据复制工作。Cassandra允许客户端控制如何复制数据。Cassandra提供了一些复制策略给客户端，比如 “RackUnaware” “Rack Aware” (在一个数据中心内) 以及 “Datacenter Aware” .应用程序通过复制策略来选择副本。如果应用程序端选择了”Rack Unaware”策略，那么系统会选择调度节点的N-1个后续节点，作为副本节点。对于”Rack Aware” 和 “Datacenter Aware” 策略算法上会复杂一些。Cassandra将会向Zookeeper做一次系统请求，获取一个领袖节点。在每个节点加入集群时候，都会向领袖节点去查询副本节点的覆盖范围，领袖节点能够确保，环中的每个节点的副本节点数量，不超过N-1. 每个节点都会本地缓存一份关于节点覆盖范围的meta数据信息，同时考虑到容灾的需求，在ZooKeeper上面也会存储一份。这样当节点崩溃时候，就会有关于这个节点覆盖范围的备份信息存在。我们借用Dynamo parlance系统中的概念，将负责节点的覆盖范围，视为优先的覆盖范围。
在5.1已经谈到，每个节点都会关注系统中其他的节点，当然也会关注节点覆盖范围之
内的节点。Cassandra在节点失效，节点间网络中断的情况下，通过降低对Quorum的要求，提供了持续服务的保证。 数据中心在电力中断，网络中断，冷却系统故障，或者自然灾害等情况下，都会失效。Cassandra可以配置成每一行多个数据中心都有副本。实际上，一个KEY的优先覆盖范围列表在构建的时候，会考虑到存储节点跨越多个数据中心的情况。这些数据中心通过高速专线网络相连。通过跨越数据中心的复制方案，我们可以处理任何数据中心的问题。
5.3节点关系 (Membership)
Cassandra中集群的节点关系依赖Scuttlebutt, Scuttlebutt基于高效的反熵GOSSIP协议。Scuttlebutt最突出的特征，是他具有高效的CPU，gossip通道利用率。在Cassandra系统中，Gossip协议不仅用来做节点关系管理，也用来传输系统相关的控制状态。
5.3.1Failure Detection
失效检测，是一种机制，通过它节点可以获取其他节点是不是在正常工作。在Cassandra中，失效检测还用来避免节点在一些操作中，同一些不可到达节点的通讯。 Cassandra采用修改过的Accrual Failure Detector. Accrual 模块不会返回一个Boolean值来标识节点是工作还是宕机状态，相反，这个模块会返回每个受监控节点的一个评估的等级，用Φ来表示，这样做的目的是Φ实际表示的一个范围，这个范围可以动态调整以反映被监控节点的网络和负载情况。
下面具体解释一下 Φ 的含义: 给定一个 Φ 的临界值，然后假定我们在 Φ=1的时候，认为A节点有问题，我们这个猜测的错误（我们的结论，可能被心跳线等其他的状态信息所推翻）概率为10% ，那么当Φ=2的时候，我们猜错的概率只有1% ，在Φ=3的时候，我们猜错的概率为0.1% … 在系统中每个节点都维护着一个其他节点发出的gossip消息的内部到达时间的滑动窗口， 系统会计算这些到达时间的分布，然后计算Φ的值。尽管原始的论文中建议通过高斯分布来拟合数据，但是我们发现根据gossip 通道的特点和他对于延迟的影响，指数分布会有更高的拟合精度。据我们所知，我们是最先采用上述方式来使用基于gossip 的 Accrual Failure Detection。 Accrual Failure Detectors 在速度和精度上都表现良好，经调整，在网络状况和服务器负载情况检查上，也有尚佳表现。
5.4启动 Bootstrapping
当一个节点最先加入到集群中时，系统会给他在环中，随机分配一个位置。考虑到容灾需求，这个映射关系会在节点本地和Zookeeper中，都做存储。然后系统会在集群中通过gossip协议广播这个位置信息。然后环中所有的节点，都知道了这个信息。这就保证了任何节点都能将key路由到其他正确的节点上面。当一个节点准备加入到集群的时候，他会读一个配置文件，配置文件中包含一些集群中可以联系的节点。我们将这些初始的联系节点，称之为集群种子。集群种子也可以通过Zookeeper配置服务来提供。
Facebook的环境中，节点离线(因为失效或者维护任务)经常瞬间完成，但是也可能持续一段时间。失效可以以多种形态出现，比如磁盘错误，CPU损坏。一般节点都是临时离线，因此不需要数据的从新分布或者修复不可达的副本。相似的，因为不小心启动了一个新的Cassandra节点，会导致人为的错误，这会让在每个Cassandra实例中的每个消息中，都会包括节点的名字。假如一个人为的配置错误让一个节点加入了错误的Cassandra实例中，会导致集群名字失效。因为这些原因，因此需要一种明确的机制，来向Cassandra实例中增加或者移除节点。管理员可以通过命令行工具或者浏览器连接到Cassandra节点上面，进行节点的增加与删除操作。
5.5集群扩容 (Scaling the Cluster )
当一个新的节点加入到系统中的时候，系统在环中为他分配一个位置，这样他就可以缓解一些节点的过重的负担。新的节点会承担一些其他节点的职能。不管是通过命令行还是web界面增加节点，系统都会执行初始化算法。其他的节点会通过内存拷贝技术，将数据传输到新的节点。根据运维经验，单节点数据传输速度能达到40M/s。我们目前在研究类似于Bittorrent多副本传输技术，通过多个副本给上线节点传输数据，从而加快节点启动过程。
5.6本地持久化 (Local Persistence)
Cassandra系统依赖本地文件系统做数据持久化。存储结构为了更有效的获取数据而设计。一般写操作包括两个步骤，先写到提交日志里面，这样可以保证系统持续服务和系统故障时候，数据可以恢复。系统会在提交日志文件成功之后，在更新内存中的数据结构。在每个节点上面，为了达到最高的数据吞吐量，都会采用一块独立的硬盘写数据提交日志。当发现内存中对象数目和数据大小达到一定的阀值的时候，数据会被dump到磁盘上面。节点都会安装多块普通硬盘，每次写操作，会写到一块指定硬盘。所有的写操作都顺序进行，并且会建立基于ROW KEY的索引，以便于快速查找。这些回写的索引，会像数据文件一样存储。当这种类型的文件多到一定程度，在后台会启动一个合并进程，将这些文件合并成一个文件。在BigTable中也有类似的合并操作。
一个典型的读操作，会先在内存中的数据结构做查找，如果内存未命中，再去做文件系统查找。做文件查找的时候，会按照由新到老的顺序。在做磁盘文件系统查找的时候，我们判断key是否在一些文件中存在。为了加快效率，如果一个文件没有包含key，那么系统就不会扫描这个文件。为此，对于每个文件的所包含的key信息，做摘要，然后写到内存里面，先采用布隆过滤器直接过滤内存。如果一个key指向了一个含有多个列的列族，那么我在做基于这个key的读取操作的时候，还需要额外的索引机制来获取列，这时单纯的key索引已经不能满足需求。为了防止扫描磁盘上每个列，我们维护了一个列的索引，这样我们可以直接去磁盘的指定块获取这个列数据。因为key索引的列 会序列化到磁盘上面，所以我们以256k为一块， 这个值也可以配置，不过我们发现对于生产环境的负载，256k已经能够满足需求。
5.7实现细节 (Implementation Details)
单机Cassandra进程，由以下部分组成： 分区模块，集群节点关系管理和失效检测模块，存储引擎模块。 这些模块依赖于事件驱动，消息处理管道和任务管道依据SEDA的架构原则，切分成多个Stage. 这些模块都由JAVA编写。集群节点关系管理和失效检测模块构建在网络层之上，采用非阻塞I/O.所有的系统控制消息基于UDP协议传输，所有的应用程序相关的消息，比如复制和请求路由，基于TCP协议传输。请求路由模块，通过一个状态机实现。当一个读写请求到达集群中的一个节点时，状态机会在如下状态间切换 (i) 确认这个拥有这个key数据的节点 (ii) 将请求路由到这些节点，并且等待请求到达的回复。(iii)如在请求在指定的超时时间内，没有回复,那么将这个请求设定成失败，并且返回客户端 (iv)根据返回的时间戳，找出最新的响应。(v) 如果发现有副本中的数据不是最新的，那么安排一个数据修复操作。本文不详细讨论失效处理的场景。系统可以配置成同步写，也可以配置成异步复制。对于需要高吞吐量的系统，我们会配置成异步写，这种系统一般为写密集型。对于同步复制的系统，在我们给客户端返回之前，我们要等待响应的节点数量达到一个最低有效值。
在任何日志型文件系统里面，都需要一种机制，来清理提交日志。当老的日志超过一定的尺寸的时候，会自动开启一个新的日志文件。 日志轮询的尺寸，可以配置，我们经验是在生产环境中，日志文件保持在128M，是个不错的选择。每一条提交日志，都有一个位向量组成的头，这个头固定大小，但是能够容纳系统所能处理的列族的最大数目。在我们的实现中，每生成一个列族，在内存和文件系统都会生成一份数据结构。每次内存中一个特定列族的数据结构成功回写到磁盘的时候，我们更改提交日志的头，将这个列族对应的位向量置位，这标志着这片信息被成功提交。每一条提交日志都会有位向量，位向量在内存中维护。当日志文件需要做轮询的时候，系统会做位向量的对比，如果当发现所有的数据都已经被持久化到磁盘上面，那么老的日志文件会被删除。写提交日志的操作，可以采用普通模式或者快速同步模式。在快速同步模式下写日志，系统会先缓冲写请求。这样做如果机器宕机，会有丢失数据的风险。Cassandra在做内存数据回写的时候，仍旧采用缓冲的方式持久化数据。传统的数据库不是为了高写吞吐量设计的，Cassandra将所有的写请求顺序的写入磁盘，以达到最高的写吞吐量。因为文件回写磁盘的时候是顺序进行的，因此不存在互斥问题，当读的时候也不需要锁。Cassandra对于读写，都是无锁状态，因此不需要处理基于B-TREE数据库的并发问题。
Cassandra根据主键索引所有的数据。磁盘中的数据文件被打碎成一些列的块，每块最多含有128个key,块之间通过块索引分割开来。块索引中包括key的偏移量和key对应的数据的大小。当内存中的数据结构回写硬盘的时候，会生成块索引，他们的偏移量会写以独立索引的形式写到独立的硬盘上面。为了访问速度，这个索引也会在内存中存一份。一般读操作都会先在内存中查找数据结构。如果在内存中命中，那么直接返回客户端，因为内存中总是存储最新的数据。如果基于内存的查找失败，那么在文件系统按照时间倒序做查找。因为我们最常查找最新的数据，因此我们如果在最新的数据文件中发现命中，那么直接返回数据。随着时间的推移，硬盘上面文件会越来越多。然后我们会像Big Table一样启动一个文件合并进程，将多个文件合并成一个。一般合并的文件，都是有序的，因此合并的文件尺寸会比较相近，比如不能将一个100G的文件同一个小于50G的文件合并。一个合并进程，会定时的将相关的文件合并成一个大文件。合并操作会是I/O密集型，因此为了不影响读，系统会采取很多优化措施。
6.实践经验 (PRACTIAL EXPERIENCES )
在设计，实现，维护Cassandra过程中，我们收获了很多。最重要的经验就是，如果没有了解到应用端对新特性的使用造成的影响，那么最好不要增加这个新特性。大部分问题不是因为节点崩溃或者网络分割引起的。 下面我们分享一下这些问题场景。
-
在Inbox Search 上线以前，我们有1亿用户，需要索引的数据有7Tb ，然后将索 引存储在Mysql里面，加载到Cassandra中。整个过程包扩在Mysql数据上面运行Map/Reduce 任务，索引他们，然后在Cassandra中存储反向索引信息。M/R 处理过程，作为Cassandra的一个客户端来执行。我们暴露了给M/R过程一些后端通道，将每个用户的倒排索引信息聚合起来，序列化，发送给Cassandra进程。这种工作方式系统的唯一瓶颈是网络带宽。
-
多数的应用程序，要求在副本进行上面Key操作为原子操作。 当然有些应用对于事物的要求，主要是为了维护一些二级指标。多数RDBMS’s 开发这都会觉得这是个有用的特性。我们在努力建立一种机制，实现这些原子操作。
-
我们也测试了其他的失效检测器，比如在附录【15】，【5】中列出的方案。我们发现，随着节点规模的扩大，失效检测时间很快超出了可以接受的范围，在一个实验中，集群有100个基点，检测出一个失效节点，花费了2分钟，这个时间对我们来讲完全不可接受。最后测试Accrual failure detector 时候，将PHI保守的设定为5，100个节点的集群，平均失效检测时间在15秒左右。
-
监控是不能想当然来做的。Cassandra 内置了分布式性能监控工具 Ganglia .我们为Ganglia提供了多样的系统级别监控指标，方便我们更好的了解在生产环境的负载情况下，我们的系统表现。 当硬盘突然失效的时候，会触发节点启动算法进行节点修复。
-
尽管Cassandra是个完全去中心化的系统，但是我们在实现某写分布式功能的时候，发现设置一些协调节点，能够让我们更容易的驾驭系统。比如Cassandra设置了Zookeeper ，在规模的集群中做协调调度工作。我们的目标是把Cassandra中同存储无关的功能，都整合到Zookeeper上面。
6.1Facebook 收件箱搜索 (Index Search)
在Facebook的收件箱搜索中，所有的发送，接收消息，都按照用户做了索引，每个用户一份。目前搜索支持两种方式(a) term search (b) 交互式搜索 – 给定一个人的名字，返回这个人所有收发的消息。对于(a)查询，用户id是key，消息存放在超级列里面，每个包含关键词的消息标识存在列里面。对于(b)查询，用户id仍旧是key,收件人的id是超级列族。对于每个超级列族，每个消息的标识都放到列里面。为了加速搜索过程，Cassandra提供了智能缓存机制。比如当一个用户点击搜索条的时候，一个异步的消息发送到Cassandra集群，集群根据这个消息准备一份这个用户的索引放到cache里面。这样当用户开始搜索的时候，搜索结果很可能已经在内存里面了。收件箱搜索系统目前运行在150个节点的集群上面，数据量50+TB。 集群节点分布在美国东海岸和西海岸的数据中心。下面的图是一些生产环境的性能数据。
Latency Stat
Search Interactions
Term Search
Min
7.69ms
7.78ms
Median
15.69ms
18.27ms
Max
26.13ms
44.41ms
7.结论 CONLUSION
我们设计，实施，运营了一个能够提供扩展性，高性能，广泛适用的存储系统。Cassandra可以在提供非常高的数据更新吞吐量时候保持低延时。未来的工作，会增加压缩，多个key的原子操作和 二级索引支持。
8.致谢 ACKNOWLEDGEMENTS
Cassandra 从内部员工那里得到了很多改进反馈。在第一次部署的时候，KarthikRanganathan 索引了所有Mysql的数据，并且帮我们迁移到Cassandra里面。 另外感谢EPFL的Dan Dumitriu 的有很多价值建议。
