分布式系统之数据分片

1、分布式系统之数据分片分布式系统（尤其是分布式存储系统）需要解决的两个最主要的问题，即数据分片和数 据冗余，下面这个图片（来源）形象生动的解释了其概念和区别：PartitionedReplicatedMode 1PartitionedReplicatedMode 1Node 2其中数据即A、B属于数据分片，原始数据被拆分成两个正交子集分布在两个节点上。 而数据集C属于数据冗余，同一份完整的数据在两个节点都有存储。当然，在实际的分布 式系统中，数据分片和数据冗余一般都是共存的。本文主要讨论数据分片的三个问题：（1）如何做数据分片，即如何将数据映射到节点（2）数据分片的特征值，即按照数据中的哪一个属

2、性（字段）来分片（3）数据分片的元数据的管理，如何保证元数据服务器的高性能、高可用，如果是一 组服务器，如何保证强一致性。所谓分布式系统，就是利用多个独立的计算机来解决单个节点（计算机）无法处理的 存储、计算问题，这是非常典型的分而治之的思想。每个节点只负责原问题（即整个系统 需要完成的任务）的一个子集，那么原问题如何拆分到多个节点？在分布式存储系统中，任 务的拆分即数据分片。何为数据分片（segment，fragment， shard，partition），就是按照一定的规则，将 数据集划分成相互独立、正交的数据子集，然后将数据子集分布到不同的节点上。注意， 这里提到，数据分片需要按照一定的

3、规则，不同的分布式应用有不同的规则，但都遵循同样 的原则：按照最主要、最频繁使用的访问方式来分片。三种数据分片方式首先介绍三种分片方式:hash方式,一致性hash （consistent hash）,按照数据范围（range based）。对于任何方式，都需要思考以下几个问题：.具体如何划分原始数据集？.当原问题的规模变大的时候，能否通过增加节点来动态适应？.当某个节点故障的时候，能否将该节点上的任务均衡的分摊到其他节点？.对于可修改的数据（比如数据库数据），如果某节点数据量变大，能否以及如何将部 分数据迁移到其他负载较小的节点，及达到动态均衡的效果？.元数据的管理（即数据与物理节点的对应关

4、系）规模？元数据更新的频率以及复杂 度？为了后面分析不同的数据分片方式，假设有三个物理节点，编号为N0，N1， N2；有 以下几条记录：R0: id: 95, name: aa, tag:olderR1: id: 302, name: bb,R2: id: 759, name: aa,R3: id: 607, name: dd, age: 18R4: id: 904, name: ff,R5: id: 246, name: gg,R6: id: 148, name: ff,R7: id: 533, name: kk, hash方式：哈希表（散列表）是最为常见的数据结构，根据记录（或者对象）的关

5、键值将记录映 射到表中的一个槽（slot），便于快速访问。绝大多数编程语言都有对hash表的支持，如 python中的dict， C+中的map，Java中的Hashtable， Lua中的table等等。在哈希表中， 最为简单的散列函数是mod N（N为表的大小）。即首先将关键值计算出hash值（这里是 一个整型），通过对N取余，余数即在表中的位置。数据分片的hash方式也是这个思想，即按照数据的某一特征保。丫）来计算哈希值，并 将哈希值与系统中的节点建立映射关系,从而将哈希值不同的数据分布到不同的节点上。我们选择id作为数据分片的key，那么各个节点负责的数据如下:由此可以看到，按照has

6、h方式做数据分片，映射关系非常简单；需要管理的元数据也 非常之少，只需要记录节点的数目以及hash方式就行了。但hash方式的缺点也非常明显：当加入或者删除一个节点的时候，大量的数据需要移 动。比如在这里增加一个节点N3,因此hash方式变为了 mod 4,数据的迁移如下：R4(id:904)R6(id:148)R7(id:533)Rl(id:302)R:5iid:246JR4(id:904)R6(id:148)R7(id:533)Rl(id:302)R:5iid:246JRO(id:95)、R2i.id：7.S9)R3iid:607)Nod el在这种方式下，是不满足单调性(Monotoni

7、city)的：如果已经有一些内容通过哈希分 派到了相应的缓冲中，又有新的缓冲加入到系统中。哈希的结果应能够保证原有已分配的内 容可以被映射到原有的或者新的缓冲中去，而不会被映射到旧的缓冲集合中的其他缓冲区。在工程中，为了减少迁移的数据量，节点的数目可以成倍增长，这样概率上来讲至多有 50%的数据迁移。hash方式还有一个缺点，即很难解决数据不均衡的问题。有两种情况：原始数据的特 征值分布不均匀，导致大量的数据集中到一个物理节点上；第二，对于可修改的记录数据， 单条记录的数据变大。在这两种情况下，都会导致节点之间的负载不均衡，而且在hash方 式下很难解决。一致性hash一致性hash是将数据按

8、照特征值映射到一个首尾相接的hash环上，同时也将节点(按 照IP地址或者机器名hash)映射到这个环上。对于数据，从数据在环上的位置开始，顺时 针找到的第一个节点即为数据的存储节点。这里仍然以上述的数据为例，假设id的范围为 0, 10001，N0, N1, N2在环上的位置分别是100, 400, 800,那么hash环示意图 与数据的分布如下：9P4：R4148:穹 6EDGModea&33:R7RO(id:95) R1(id:904)302：iRlNoclc?0R6(icl: 148)RI(id:302) R5 Gd: 246)XodclR2(id:759)R3(id:607)9P4：

9、R4148:穹 6EDGModea&33:R7RO(id:95) R1(id:904)302：iRlNoclc?0R6(icl: 148)RI(id:302) R5 Gd: 246)XodclR2(id:759)R3(id:607)R7(id:5：n)Nod 24,ea7：M |7 的;F121OO；N(MJe096: HO24S；R5可以看到相比于上述的hash方式，一致性hash方式需要维护的元数据额外包含了节点 在环上的位置，但这个数据量也是非常小的。一致性hash在增加或者删除节点的时候，受到影响的数据是比较有限的，比如这里增 加一个节点N3,其在环上的位置为600，因此，原来N2负责

10、的范围段(400，800现在 由 N2 (400, 600 N3(600, 800负责，因此只需要将记录 R2(id:759), R3(id: 607)从 N2,迁移 到N3:不难发现一致性hash方式在增删的时候只会影响到hash环上响应的节点，不会发生 大规模的数据迁移。但是，一致性hash方式在增加节点的时候，只能分摊一个已存在节点的压力；同样， 在其中一个节点挂掉的时候，该节点的压力也会被全部转移到下一个节点。我们希望的是“一 方有难，八方支援”，因此需要在增删节点的时候，已存在的所有节点都能参与响应，达到 新的均衡状态。因此，在实际工程中，一般会引入虚拟节点Virtual node)

11、的概念。艮坏是将物理节点 映射在hash换上，而是将虚拟节点映射到hash环上。虚拟节点的数目远大于物理节点，因 此一个物理节点需要负责多个虚拟节点的真实存储。操作数据的时候，先通过hash环找到 对应的虚拟节点，再通过虚拟节点与物理节点的映射关系找到对应的物理节点。引入虚拟节点后的一致性hash需要维护的元数据也会增加：第一，虚拟节点在hash环 上的问题，且虚拟节点的数目又比较多；第二，虚拟节点与物理节点的映射关系。但带来 的好处是明显的，当一个物理节点失效是，hash环上多个虚拟节点失效，对应的压力也就 会发散到多个其余的虚拟节点，事实上也就是多个其余的物理节点。在增加物理节点的时候 同

12、样如此。工程中，Dynamo、Cassandra都使用了一致性hash算法，且在比较高的版本中都使用 了虚拟节点的概念。在这些系统中，需要考虑综合考虑数据分布方式和数据副本，当引入数据副本之后，一致性hash方式也需要做相应的调整，可以参加cassandra的相关文档。range based简单来说，就是按照关键值划分成不同的区间，每个物理节点负责一个或者多个区间。 其实这种方式跟一致性hash有点像，可以理解为物理节点在hash环上的位置是动态变化的。RO (id； 96)R6(id:148)R7(id:533)RI (1d:302)R5(id:246)还是以上面的数据举例，三个节点的数据区

13、间分别是N0（0, 200，N1（200, 500RO (id； 96)R6(id:148)R7(id:533)RI (1d:302)R5(id:246)K2 (id: 759)R3(id:607)R4 Gdi 904)NcdrON(idelndc2NcdrON(idelndc2注意，区间的大小不是固定的，每个数据区间的数据量与区间的大小也是没有关系的。 比如说，一部分数据非常集中，那么区间大小应该是比较小的，即以数据量的大小为片段 标准。在实际工程中，一个节点往往负责多个区间，每个区间成为一个块（chunk、block）， 每个块有一个阈值，当达到这个阈值之后就会分裂成两个块。这样做的目的在

14、于当有节点 加入的时候，可以快速达到均衡的目的。不知道读者有没有发现，如果一个节点负责的数据只有一个区间，range based与没有 虚拟节点概念的一致性hash很类似；如果一个节点负责多个区间，range based与有虚拟节 点概念的一致性hash很类似。range based的元数据管理相对复杂一些，需要记录每个节点的数据区间范围，特别单 个节点对于多个区间的情况。而且，在数据可修改的情况下，如果块进行分裂，那么元数据 中的区间信息也需要同步修改。range based这种数据分片方式应用非常广泛，比如MongoDB, PostgreSQL， HDFS 小结：在这里对三种分片方式（应该

15、是四种，有没有virtual node的一致性hash算两种）进行 简单总结，主要是针对提出的几个问题：映射难度元数据节点增删hash方式简单非常简单，几 乎不用修改需要迁移的数据比较多consistent hash without virtual node简单比较简单，取 决于节点规 模，几乎不用 修改增删节点的时候只影响hash环上相邻节点， 但不能使所有节点都参与数据迁移过程consistent hash with virtual node中等稍微复杂一 些，主要取决 于虚拟节点规 模，很少修改需要迁移的数据比较少，且所有节点都能贡献部分数据range较为复杂取决于每个块需要迁移的数据比

16、较少，且所有节点都能贡献部分数据based的大小，一般来说规模较大；且修改频率较高上面的数据动态均衡，值得是上述问题的第4点，即如果某节点数据量变大，能否以及 如何将部分数据迁移到其他负载较小的节点.分片特征值的选择上面的三种方式都提到了对数据的分片是基于关键值、特征值的。这个特征值在不同的 系统中有不同的叫法，比如MongoDB中的sharding key，Oracle中的Partition Key，不管怎 么样，这个特征值的选择都是非常非常重要的。那么。怎么选择这个特征值呢？Distributed systems for fun and profit给出了言简意 赅的标准：based on

17、 what you think the primary access pattern will be.大概翻译为：基于最常用的访问模式。访问时包括对数据的增删改查的。比如上面的列 子，我们选择“id”作为分片的依据，那么就是默认对的数据增删改查都是通过“id”字段 来进行的。如果在应用中，大量的数据操作都是通过这个特征值进行，那么数据分片就能提供两个 额外的好处：（1）提升性能和并发，操作被分发到不同的分片，相互独立.（2）提升系统的可用性，即使部分分片不能用，其他分片不会受到影响.如果大量操作并没有使用到特征值，那么就很麻烦了。比如在本文的例子中，如果用 name去查询，而元数据记录的是如何

18、根据按照id映射数据位置，那就尴尬了，需要到多有 分片都去查一下，然后再做一个聚合！另外一个问题，如果以单个字段为特征值（如id），那么不管按照什么分布方式，在多 条数据拥有相同的特征值（如id）的情况下，这些数据一定都会分布到同一个节点上。在这 种情况下有两个问题，一是不能达到节点间数据的均衡，二是如果数据超过了单个节点的存 储能力怎么办？关键在于，即使按照分布式系统解决问题的常规办法-增加节点-也是于 事无补的。在这个时候，单个字段做特征值就不行了，可能得再增加一个字段作为“联合特征值”， 类似数据库中的联合索引。比如，数据是用户的操作日志，可以使用id和时间戳一起作为 hash函数的输入

19、，然后算出特征值；但在这种情况下，如果还想以id为查询关键字来查询， 那就得遍历所有节点了。所以说没有最优的设计，只有最符合应用需求的设计。下面以MongoDB中的sharding key为例，解释特征值选择的重要性以及对数据操作的 影响。如果有数据库操作基础，即使没有使用过MongoDB，阅读下面的内容应该也没有问 题。以 MongoDB sharding key 为例关于MongoDB Sharded cluster，之前也写过一篇文章通过一步步创建sharded cluster 来认识mongodb，做了简单介绍。在我的工作场景中，除了联合查询（join）和事务，MongoDB 的使用和

20、Mysql还是比较相似的，特别是基本的CRUD操作、数据库索引。MongoDb中， 每一个分片成为一个shard，分片的特征值成为sharding key，每个数据称之为一个document。 选择适合的字段作为shardingkey非常重要，why？前面也提到，如果使用非sharding key去访问数据，那么元数据服务器（或者元数据缓 存服务器，后面会讲解这一部分）是没法知道对应的数据在哪一个shard上，那么该访问就 得发送到所有的shard，得到所有shard的结果之后再做聚合，在mongoDB中，由mongos （缓存有元数据信息）做数据聚合。对于数据读取（R： read or re

21、trieve），通过同一个字 段获取到多个数据，是没有问题的，只是效率比较低而已。对于数据更新，如果只能更新一 个数据，那么在哪一个shard上更新呢，似乎都不对，这个时候，MongoDB是拒绝的。对 应到MongoDB（MongoDD3.0）的命令包括但不限于：findandmodify：这个命令只能更新一个document，因此查询部分必须包含sharding key When using findAndModify in a sharded environment, the query must contain the shard key for all operations again

22、st the shard cluster for the sharded collections.update：这个命令有一个参数multi，默认是false，即只能更新一个document，此时查 询部分必须包含sharding keyAll update（） operations for a sharded collection that specify the multi: false option must include theshard key or the _id field in the query specification.remove：有一个参数JustOne，如果为Tr

23、ue，只能删除一个document，也必须使用 sharidng key另外，熟悉sql的同学都知道，在数据中索引中有unique index （唯一索引），即保证 这个字段的值在table中是唯一的。mongoDB中，也可以建立 unique index，但是在 sharded cluster环境下，只能对 sharding key创建unique index，道理也很简单，如果 unique index不是sharidng key，那么插入的时候就得去所有shard上查看，而且还得加锁。接下来，讨论分片到shard上的数据不均的问题，如果一段时间内shardkey过于集中（比 如按时间增长

24、），那么数据只往一个shard写入，导致无法平衡集群压力。MongoDB中提供yrange partition和hash partition，这个跟上面提到的分片方式 hash方式，ranged based不是一回事儿，而是指对sharding key处理。MongoDB 一定是ranged base分片方式,docuemnt中如是说：MongoDB partitions data in the collection using ranges of shard key values. Each range defines a non-overlapping range of shard key

25、 values and is associated with a chunk.那么什么是range partition和hash partition，官网的一张图很好说明了二者的区别：Chunk AjninKey X 10Chunk BChunk CChunk AjninKey X 10Chunk BChunk C20 X maxKeymaxKeyChunk AChunk BChunk CminKey X 10HashFunction上图左是maxKeyChunk AChunk BChunk CminKey X 10HashFunction上图左是range partition,右是 hash

26、 partition。range partition就是使用字段本身作为分 片的边界，比如上图的x；而hash partition会将字段重新hash到一个更大、更离散的值域 区间。hash partition的最大好处在于保证数据在各个节点上均匀分布(这里的均匀指的是在写 入的时候就均匀，而不是通过MongoDB的balancing功能)。比如MongoDB中默认的Jd 是objectid，objectid是一个12个字节的BSON类型，前4个字节是机器的时间戳，那么如 果在同一时间大量创建以ObjectId为Jd的数据会分配到同一个shard上，此时若将id设置 为hash index和h

27、ash sharding key，就不会有这个问题。当然，hash partition相比range partition也有一个很大的缺点，就是范围查询的时候效 率低！因此到底选用hash partition还是range partition还得根据应用场景来具体讨论。最后得知道，sharding key 一但选定，就无法修改(Immutable)。如果应用必须要修改 sharidng key，那么只能将数据导出，新建数据库并创建新的sharding key，最后导入数据。元数据服务器在上面讨论的三种数据分片分式中，或多或少都会记录一些元数据：数据与节点的映 射关系、节点状态等等。我们称记录元

28、数据的服务器为元数据服务器(metaserver)，不同 的系统叫法不一样，比如 master、configserver、namenode 等。元数据服务器就像人类的大脑，一只手不能用了还没忍受，大脑不工作整个人就瘫痪了。 因此，元数据服务器的高性能、高可用，要达到这两个目标，元数据服务器就得高可扩展- 以此应对元数据的增长。元数据的高可用要求元数据服务器不能成为故障单点(single point of failure)，因此需 要元数据服务器有多个备份，并且能够在故障的时候迅速切换。有多个备份，那么问题就来了，怎么保证多个备份的数据一致性？多个副本的一致性、可用性是CAP理论讨论的范畴，这里

29、简单介绍两种方案。第一种 是主从同步，首先选出主服务器，只有主服务器提供对外服务，主服务器将元数据的变革信 息以日志的方式持久化到共享存储(例如nfs)，然后从服务器从共享存储读取日志并应用, 达到与主服务器一致的状态，如果主服务器被检测到故障(比如通过心跳)，那么会重新选 出新的主服务器。第二种方式，通过分布式一致性协议来达到多个副本件的一致，比如大名 鼎鼎的Paxos协议，以及工程中使用较多的Paxos的特化版本-Raft协议，协议可以实现所 有备份均可以提供对外服务，并且保证强一致性。HDFS元数据HDFS中，元数据服务器被称之为namenode，在hdfs1.0之前，namenode还

30、是单点，一旦namenode挂掉，整个系统就无法工作。在hdfs2.0，解决了 namenode的单点问题。上图中NN即NameNode， DN即DataNode (即实际存储数据的节点)。从图中可以看 至U， 两台NameNode形成互备，一台处于 Active状态，为主 NameNode，另外一台处 于Standby状态，为备NameNode,只有主NameNode才能对外提供读写服务。Active NN与standby NN之间的数据同步通过共享存储实现，共享存储系统保证了 Namenode的高可用。为了保证元数据的强一致性，在进行准备切换的时候，新的Active NN 必须要在确认元数

31、据完全同步之后才能继续对外提供服务。另外，Namenode的状态监控以及准备切换都是Zookeeper集群负责，在网络分割 (network partition)的情况下，有可能zookeeper认为原来的Active NN挂掉了，选举出新 的ActiveNN,但实际上原来的Active NN还在继续提供服务。这就导致了“双主”或者脑裂 (brain-split)现象。为了解决这个问题，提出了 fencing机制，也就是想办法把 旧的Active NameNode隔离起来，使它不能正常对外提供服务。具体参见这篇文章。MongoDB元数据MongoDB中，元数据服务器被称为config serv

32、er。在MongoDB3.2中，已经不再建议使 用三个镜像(Mirrored)MongoDB实例作为config server,而是推荐使用复制集(replica set) 作为config server,此举的目的是增强config server的一致性，而且config sever中mongod 的数目也能从3个达到replica set的上线(50个节点)，从而提高了可靠性。在MongoDB3.0及之前的版本中，元数据的读写按照下面的方式进行：When writing to the three config servers, a coordinator dispatches the sa

33、me write commands to the three config servers and collects the results. Differing results indicate an inconsistent writes to the config servers and may require manual intervention.MongoDB的官方文档并没有详细解释这一过程，不过在stackexchange上，有人指出这 个过程是两阶段提交。MongoDB3.2 及之后的版本，使用 了 replica set config server,在CAP 理论与 Mong

34、oDB 一致性、可用性的一些思考文章中，详细介绍yreplica set的write concern、read concern 和read references,这三个选项会影响到复制集的一致性、可靠性与读取性能。在config server 中，使用 了 WriteConcern： Majority； ReadConcern： Majority； ReadReferences： nearest。元数据的缓存：即使元数据服务器可以由一组物理机器组成，也保证了副本集之间的一致性问题。但是 如果每次对数据的请求都经过元数据服务器的话，元数据服务器的压力也是非常大的。很多 应用场景，元数据的变化并不

35、是很频繁，因此可以在访问节点上做缓存，这样应用可以直接 利用缓存数据进行数据读写，减轻元数据服务器压力。在这个环境下，缓存的元数据必须与元数据服务器上的元数据一致，缓存的元数据必须 是准确的，未过时的。相反的例子是DNS之类的缓存，即使使用了过期的DNS缓存也不会 有太大的问题。怎么达到缓存的强一致性呢？比较容易想到的办法是当metadata变化的时候立即通知 所有的缓存服务器（mongos）,但问题是通信有延时，不可靠。解决不一致的问题，一个比较常见的思路是版本号，比如网络通信，通信协议可能会发 生变化，通信双方为了达成一致，那么可以使用版本号。在缓存一致性的问题上，也可以使 用版本号，基本

36、思路是请求的时候带上缓存的版本号，路由到具体节点之后比较实际数据的 版本号，如果版本号不一致，那么表示缓存信息过旧，此时需要从元数据服务器重新拉取元 数据并缓存。在MongoDB中，mongos缓存上就是使用的这种办法。另外一种解决办法，就是大名鼎鼎的lease机制-“An Efficient Fault-Tolerant Mechanism for Distributed File Cache Consistency”, lease机制在分布式系统中使用非常广泛，不仅仅用 于分布式缓存，在很多需要达成某种约定的地方都大显身手，在分布式系统原理介绍中， 对lease机制有较为详细的描述，下面对

37、lease机制进行简单介绍。Lease机制：既然，Lease机制提出的时候是为了解决分布式存储系统中缓存一致性的问题，那么首 先来看看Lease机制是怎么保证缓存的强一致性的。注意，为了方便后文描述，在本小节中， 我们称元数据服务器为服务器，缓存服务器为客户端。要点：服务器向所有客户端发送缓存数据的同时，颁发一个lease, lease包含一个有限期（即 过期时间）lease的含义是：在这个有效期内，服务器保证元数据不会发生变化因此客户端在这个有效期内可以放心大胆的使用缓存的元数据，如果超过了有效期，就 不能使用数据了，就得去服务器请求。如果外部请求修改服务器上的元数据（元数据的修改一定在服务器上进行），那么服务 器会阻塞修改请求，直到所有已颁发的lease过期，然后修改元数据，并将新的元数据和新 的lease发送到客户端如果元数据没有发生变化，那么服务器也需要在之前已颁发的lease到期之间，重新给 客户端颁发新的leas