带着问题学习分布式系统之数据分片
Posted xybaby
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了带着问题学习分布式系统之数据分片相关的知识，希望对你有一定的参考价值。
在前文中，提出了分布式系统（尤其是分布式存储系统）需要解决的两个最主要的问题，即数据分片和数据冗余，下面这个图片（来源）形象生动的解释了其概念和区别：
其中，数据集A、B属于数据分片，原始数据被拆分成两个正交子集分布在两个节点上。而数据集C属于数据冗余，同一份完整的数据在两个节点都有存储。当然，在实际的分布式系统中，数据分片和数据冗余一般都是共存的。
本文主要讨论数据分片的三个问题：
（1）如何做数据分片，即如何将数据映射到节点
（2）数据分片的特征值，即按照数据中的哪一个属性（字段）来分片
（3）数据分片的元数据的管理，如何保证元数据服务器的高性能、高可用，如果是一组服务器，如何保证强一致性
所谓分布式系统，就是利用多个独立的计算机来解决单个节点（计算机）无法处理的存储、计算问题，这是非常典型的分而治之的思想。每个节点只负责原问题（即整个系统需要完成的任务）的一个子集，那么原问题如何拆分到多个节点？在分布式存储系统中，任务的拆分即数据分片。
何为数据分片（segment，fragment， shard， partition），就是按照一定的规则，将数据集划分成相互独立、正交的数据子集，然后将数据子集分布到不同的节点上。注意，这里提到，数据分片需要按照一定的规则，不同的分布式应用有不同的规则，但都遵循同样的原则：按照最主要、最频繁使用的访问方式来分片。
本文地址：http://www.cnblogs.com/xybaby/p/7076731.html
三种数据分片方式
- 具体如何划分原始数据集？
- 当原问题的规模变大的时候，能否通过增加节点来动态适应？
- 当某个节点故障的时候，能否将该节点上的任务均衡的分摊到其他节点？
- 对于可修改的数据（比如数据库数据），如果某节点数据量变大，能否以及如何将部分数据迁移到其他负载较小的节点，及达到动态均衡的效果？
- 元数据的管理（即数据与物理节点的对应关系）规模？元数据更新的频率以及复杂度？
为了后面分析不同的数据分片方式，假设有三个物理节点，编号为N0， N1， N2；有以下几条记录：
R0: {id: 95, name: \'aa\', tag:\'older\'}
R1: {id: 302, name: \'bb\',}
R2: {id: 759, name: \'aa\',}
R3: {id: 607, name: \'dd\', age: 18}
R4: {id: 904, name: \'ff\',}
R5: {id: 246, name: \'gg\',}
R6: {id: 148, name: \'ff\',}
R7: {id: 533, name: \'kk\',}
hash方式：
哈希表（散列表）是最为常见的数据结构，根据记录（或者对象）的关键值将记录映射到表中的一个槽（slot），便于快速访问。绝大多数编程语言都有对hash表的支持，如python中的dict， C++中的map，Java中的Hashtable， Lua中的table等等。在哈希表中，最为简单的散列函数是 mod N（N为表的大小）。即首先将关键值计算出hash值（这里是一个整型），通过对N取余，余数即在表中的位置。
数据分片的hash方式也是这个思想，即按照数据的某一特征（key）来计算哈希值，并将哈希值与系统中的节点建立映射关系,从而将哈希值不同的数据分布到不同的节点上。
我们选择id作为数据分片的key，那么各个节点负责的数据如下：
由此可以看到，按照hash方式做数据分片，映射关系非常简单；需要管理的元数据也非常之少，只需要记录节点的数目以及hash方式就行了。
在这种方式下，是不满足单调性（Monotonicity）的：如果已经有一些内容通过哈希分派到了相应的缓冲中，又有新的缓冲加入到系统中，哈希的结果应能够保证原有已分配的内容可以被映射到原有的或者新的缓冲中去，而不会被映射到旧的缓冲集合中的其他缓冲区。
在工程中，为了减少迁移的数据量，节点的数目可以成倍增长，这样概率上来讲至多有50%的数据迁移。
一致性hash
可以看到相比于上述的hash方式，一致性hash方式需要维护的元数据额外包含了节点在环上的位置，但这个数据量也是非常小的。
一致性hash在增加或者删除节点的时候，受到影响的数据是比较有限的，比如这里增加一个节点N3，其在环上的位置为600，因此，原来N2负责的范围段（400， 800］现在由N3（400， 600] N2(600, 800]负责，因此只需要将记录R7(id:533) 从N2,迁移到N3:
但是，一致性hash方式在增加节点的时候，只能分摊一个已存在节点的压力；同样，在其中一个节点挂掉的时候，该节点的压力也会被全部转移到下一个节点。我们希望的是“一方有难，八方支援”，因此需要在增删节点的时候，已存在的所有节点都能参与响应，达到新的均衡状态。
因此，在实际工程中，一般会引入虚拟节点（virtual node）的概念。即不是将物理节点映射在hash换上，而是将虚拟节点映射到hash环上。虚拟节点的数目远大于物理节点，因此一个物理节点需要负责多个虚拟节点的真实存储。操作数据的时候，先通过hash环找到对应的虚拟节点，再通过虚拟节点与物理节点的映射关系找到对应的物理节点。
引入虚拟节点后的一致性hash需要维护的元数据也会增加：第一，虚拟节点在hash环上的问题，且虚拟节点的数目又比较多；第二，虚拟节点与物理节点的映射关系。但带来的好处是明显的，当一个物理节点失效是，hash环上多个虚拟节点失效，对应的压力也就会发散到多个其余的虚拟节点，事实上也就是多个其余的物理节点。在增加物理节点的时候同样如此。除此之外，可以根据物理节点的性能来调整每一个物理节点对于虚拟节点的数量，充分、合理利用资源。
工程中，Dynamo、Cassandra都使用了一致性hash算法，且在比较高的版本中都使用了虚拟节点的概念。在这些系统中，需要考虑综合考虑数据分布方式和数据副本，当引入数据副本之后，一致性hash方式也需要做相应的调整， 可以参加cassandra的相关文档。
range based
简单来说，就是按照关键值划分成不同的区间，每个物理节点负责一个或者多个区间。其实这种方式跟一致性hash有点像，可以理解为物理节点在hash环上的位置是动态变化的。
还是以上面的数据举例，三个节点的数据区间分别是N0(0, 200]， N1(200, 500]， N2(500, 1000]。那么数据分布如下：
注意，区间的大小不是固定的，每个数据区间的数据量与区间的大小也是没有关系的。比如说，一部分数据非常集中，那么区间大小应该是比较小的，即以数据量的大小为片段标准。在实际工程中，一个节点往往负责多个区间，每个区间成为一个块（chunk、block），每个块有一个阈值，当达到这个阈值之后就会分裂成两个块。这样做的目的在于当有节点加入的时候，可以快速达到均衡的目的。
不知道读者有没有发现，如果一个节点负责的数据只有一个区间，range based与没有虚拟节点概念的一致性hash很类似；如果一个节点负责多个区间，range based与有虚拟节点概念的一致性hash很类似。
range based的元数据管理相对复杂一些，需要记录每个节点的数据区间范围，特别单个节点对于多个区间的情况。而且，在数据可修改的情况下，如果块进行分裂，那么元数据中的区间信息也需要同步修改。
range based这种数据分片方式应用非常广泛，比如MongoDB, PostgreSQL， HDFS
小结：
在这里对三种分片方式（应该是四种，有没有virtual node的一致性hash算两种）进行简单总结，主要是针对提出的几个问题：
|
|
映射难度 | 元数据 | 节点增删 | 数据动态均衡 |
| hash方式 | 简单 | 非常简单，几乎不用修改 | 需要迁移的数据比较多 | 不支持 |
| consistent hash without virtual node |
简单 | 比较简单，取决于节点规模，几乎不用修改 | 增删节点的时候只影响hash环上相邻节点，但不能使所有节点都参与数据迁移过程 | 不支持 |
| consistent hash with virtual node |
中等 | 稍微复杂一些，主要取决于虚拟节点规模，很少修改 | 需要迁移的数据比较少，且所有节点都能贡献部分数据 | 若支持（修改虚拟节点与物理节点映射关系） |
| range based | 较为复杂 | 取决于每个块的大小，一般来说规模较大；且修改频率较高 | 需要迁移的数据比较少，且所有节点都能贡献部分数据 | 支持，且比较容易 |
上面的数据动态均衡，值得是上述问题的第4点，即如果某节点数据量变大，能否以及如何将部分数据迁移到其他负载较小的节点
分片特征值的选择
上面的三种方式都提到了对数据的分片是基于关键值、特征值的。这个特征值在不同的系统中有不同的叫法，比如MongoDB中的sharding key， Oracle中的Partition Key，不管怎么样，这个特征值的选择都是非常非常重要的。
那么。怎么选择这个特征值呢？《Distributed systems for fun and profit》给出了言简意赅的标准：
based on what you think the primary access pattern will be
大概翻译为：基于最常用的访问模式。这里的访问是广义的，包括对数据的增删改查。比如上面的列子，我们选择“id”作为分片的依据，那么就是默认对的数据增删改查都是通过“id”字段来进行的。
如果在应用中，大量的数据操作都是通过这个特征值进行，那么数据分片就能提供两个额外的好处：
（1）提升性能和并发，操作被分发到不同的分片，相互独立
（2）提升系统的可用性，即使部分分片不能用，其他分片不会受到影响
如果大量操作并没有使用到特征值，那么就很麻烦了。比如在本文的例子中，如果用name去查询，而元数据记录的是如何根据按照id映射数据位置，那就尴尬了，需要到多有分片都去查一下，然后再做一个聚合！
另外一个问题，如果以单个字段为特征值（如id），那么不管按照什么分布方式，在多条数据拥有相同的特征值（如id）的情况下，这些数据一定都会分布到同一个节点上。在这种情况下有两个问题，一是不能达到节点间数据的均衡，二是如果数据超过了单个节点的存储能力怎么办？关键在于，即使按照分布式系统解决问题的常规办法 -- 增加节点 --也是于事无补的。
在这个时候，单个字段做特征值就不行了，可能得再增加一个字段作为“联合特征值”，类似数据库中的联合索引。比如，数据是用户的操作日志，可以使用id和时间戳一起作为hash函数的输入，然后算出特征值；但在这种情况下，如果还想以id为查询关键字来查询，那就得遍历所有节点了。
所以说没有最优的设计，只有最符合应用需求的设计。
下面以MongoDB中的sharding key为例，解释特征值选择的重要性以及对数据操作的影响。如果有数据库操作基础，即使没有使用过MongoDB，阅读下面的内容应该也没有问题。
以MongoDB sharding key为例
关于MongoDB Sharded cluster，之前也写过一篇文章《通过一步步创建sharded cluster来认识mongodb》，做了简单介绍。在我的工作场景中，除了联合查询（join）和事务，MongoDB的使用和mysql还是比较相似的，特别是基本的CRUD操作、数据库索引。MongoDb中，每一个分片称为一个shard，分片的特征值称为sharding key，每个数据称之为一个document。选择适合的字段作为shardingkey非常重要，why？
前面也提到，如果使用非sharding key去访问数据，那么元数据服务器（或者元数据缓存服务器，后面会讲解这一部分）是没法知道对应的数据在哪一个shard上，那么该访问就得发送到所有的shard，得到所有shard的结果之后再做聚合，在mongoDB中，由路由节点mongos（缓存有元数据信息）做数据聚合。对于数据读取（R： read or retrieve），通过同一个字段获取到多个数据，是没有问题的，只是效率比较低而已。对于数据更新，如果只能更新一个数据，那么在哪一个shard上更新呢，似乎都不对，这个时候，MongoDB是拒绝的。对应到MongoDB（MongoDD3.0）的命令包括但不限于：
- findandmodify：这个命令只能更新一个document，因此查询部分必须包含sharding key
When using
findAndModifyin a sharded environment, thequerymust contain the shard key for all operations against the shard cluster for the sharded collections.
- update：这个命令有一个参数multi，默认是false，即只能更新一个document，此时查询部分必须包含sharding key
Allupdate()operations for a sharded collection that specify themulti: falseoption must include theshard key or the_idfield in the query specification.
- remove：有一个参数JustOne，如果为True，只能删除一个document，也必须使用sharidng key
另外，熟悉sql的同学都知道，在数据中索引中有unique index（唯一索引），即保证这个字段的值在table中是唯一的。mongoDB中，也可以建立unique index，但是在sharded cluster环境下，只能对sharding key创建unique index，道理也很简单，如果unique index不是sharidng key，那么插入的时候就得去所有shard上查看，而且还得加锁。
接下来，讨论分片到shard上的数据不均的问题，如果一段时间内shardkey过于集中（比如按时间增长），那么数据只往一个shard写入，导致无法平衡集群压力。
MongoDB partitions data in the collection using ranges of shard key values. Each range defines a non-overlapping range of shard key values and is associated with a chunk.
那么什么是"range partition"和"hash partition"，官网的一张图很好说明了二者的区别：
上图左是range partition，右是hash partition。range partition就是使用字段本身作为分片的边界，比如上图的x；而hash partition会将字段重新hash到一个更大、更离散的值域区间。
hash partition的最大好处在于保证数据在各个节点上的均匀分布（这里的均匀指的是在写入的时候就均匀，而不是通过MongoDB的balancing功能）。比如MongoDB中默认的_id是objectid，objectid是一个12个字节的BSON类型，前4个字节是机器的时间戳，那么如果在同一时间大量创建以ObjectId为_id的数据 会分配到同一个shard上，此时若将_id设置为hash index 和 hash sharding key，就不会有这个问题。
当然，hash partition相比range partition也有一个很大的缺点，就是范围查询的时候效率低！因此到底选用hash partition还是range partition还得根据应用场景来具体讨论。
最后得知道，sharding key一旦选定，就无法修改（Immutable）。如果应用必须要修改sharidng key，那么只能将数据导出，新建数据库并创建新的sharding key，最后导入数据。
元数据服务器
在上面讨论的三种数据分片分式中，或多或少都会记录一些元数据：数据与节点的映射关系、节点状态等等。我们称记录元数据的服务器为元数据服务器（metaserver），不同的系统叫法不一样，比如master、configserver、namenode等。
元数据服务器就像人类的大脑，一只手不能用了还没忍受，大脑不工作整个人就瘫痪了。因此，元数据服务器的高性能、高可用，要达到这两个目标，元数据服务器就得高可扩展 -- 以此应对元数据的增长。
元数据的高可用要求元数据服务器不能成为故障单点（single point of failure），因此需要元数据服务器有多个备份，并且能够在故障的时候迅速切换。
有多个备份，那么问题就来了，怎么保证多个备份的数据一致性？
多个副本的一致性、可用性是CAP理论讨论的范畴，这里简单介绍两种方案。第一种是主从同步，首先选出主服务器，只有主服务器提供对外服务，主服务器将元数据的变更信息以日志的方式持久化到共享存储（例如nfs），然后从服务器从共享存储读取日志并应用，达到与主服务器一致的状态，如果主服务器被检测到故障（比如通过心跳），那么会重新选出新的主服务器。第二种方式，通过分布式一致性协议来达到多个副本间的一致，比如大名鼎鼎的Paxos协议，以及工程中使用较多的Paxos的特化版本 -- Raft协议，协议可以实现所有备份均可以提供对外服务，并且保证强一致性。
HDFS元数据
HDFS中，元数据服务器被称之为namenode，在hdfs1.0之前，namenode还是单点，一旦namenode挂掉，整个系统就无法工作。在hdfs2.0，解决了namenode的单点问题。
上图中NN即NameNode， DN即DataNode（即实际存储数据的节点）。从图中可以看到， 两台 NameNode 形成互备，一台处于 Active 状态，为主 NameNode，另外一台处于 Standby 状态，为备 NameNode，只有主 NameNode 才能对外提供读写服务。
Active NN与standby NN之间的数据同步通过共享存储实现，共享存储系统保证了Namenode的高可用。为了保证元数据的强一致性，在进行准备切换的时候，新的Active NN必须要在确认元数据完全同步之后才能继续对外提供服务。
另外，Namenode的状态监控以及准备切换都是Zookeeper集群负责，在网络分割（network partition）的情况下，有可能zookeeper认为原来的Active NN挂掉了，选举出新的ActiveNN，但实际上原来的Active NN还在继续提供服务。这就导致了“双主“或者脑裂（brain-split）现象。为了解决这个问题，提出了fencing机制，也就是想办法把旧的 Active NameNode 隔离起来，使它不能正常对外提供服务。具体参见这篇文章。
MongoDB元数据
MongoDB中，元数据服务器被称为config server。在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 same 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理论与MongoDB一致性、可用性的一些思考》文章中，详细介绍了replica set的write concern、read concern和read references，这三个选项会影响到复制集的一致性、可靠性与读取性能。在config server中，使用了WriteConcern：Majority；ReadConcern：Majority；ReadReferences：nearest。
元数据的缓存：
即使元数据服务器可以由一组物理机器组成，也保证了副本集之间的一致性问题。但是如果每次对数据的请求都经过元数据服务器的话，元数据服务器的压力也是非常大的。很多应用场景，元数据的变化并不是很频繁，因此可以在访问节点上做缓存，这样应用可以直接利用缓存数据进行数据读写，减轻元数据服务器压力。
在这个环境下，缓存的元数据必须与元数据服务器上的元数据一致，缓存的元数据必须是准确的，未过时的。相反的例子是DNS之类的缓存，即使使用了过期的DNS缓存也不会有太大的问题。
怎么达到缓存的强一致性呢？比较容易想到的办法是当metadata变化的时候立即通知所有的缓存服务器（mongos），但问题是通信有延时，不可靠。
解决不一致的问题，一个比较常见的思路是版本号，比如网络通信，通信协议可能会发生变化，通信双方为了达成一致，那么可以使用版本号。在缓存一致性的问题上，也可以使用版本号，基本思路是请求的时候带上缓存的版本号，路由到具体节点之后比较实际数据的版本号，如果版本号不一致，那么表示缓存信息过旧，此时需要从元数据服务器重新拉取元数据并缓存。在MongoDB中，mongos缓存上就是使用的这种办法。
另外一种解决办法，就是大名鼎鼎的lease机制 -- “An Efficient Fault-Tolerant Mechanism for Distributed File Cache Consistency”，lease机制在分布式系统中使用非常广泛，不仅仅用于分布式缓存，在很多需要达成某种约定的地方都大显身手，在《分布式系统原理介绍》中，对lease机制有较为详细的描述，下面对lease机制进行简单介绍。
Lease机制：
既然，Lease机制提出的时候是为了解决分布式存储系统中缓存一致性的问题，那么首先来看看Lease机制是怎么保证缓存的强一致性的。注意，为了方便后文描述，在本小节中，我们称元数据服务器为服务器，缓存服务器为客户端。
要点：
- 服务器向所有客户端发送缓存数据的同时，颁发一个lease，lease包含一个有限期（即过期时间）
- lease的含义是：在这个有效期内，服务器保证元数据不会发生变化
- 因此客户端在这个有效期内可以放心大胆的使用缓存的元数据，如果超过了有效期，就不能使用数据了，就得去服务器请求。
- 如果外部请求修改服务器上的元数据（元数据的修改一定在服务器上进行），那么服务器会阻塞修改请求，直到所有已颁发的lease过期，然后修改元数据，并将新的元数据和新的lease发送到客户端
- 如果元数据没有发生变化，那么服务器也需要在之前已颁发的lease到期之间，重新给客户端颁发新的lease（只有lease，没有数据）
在Lease论文的标题中，提到了“Fault-Tolerant”，那么lease是怎么做到容错的呢。关键在于，只要服务器一旦发出数据和lease，不关心客户端是否收到数据，只要等待lease过期，就可以修改元数据；另外，lease的有效期通过过期时间（一个时间戳）来标识，因此即使从服务器到客户端的消息延时到达、或者重复发送都是没有关系的。
不难发现，容错的前提是服务器与客户端的时间要一致。如果服务器的时间比客户端的时间慢，那么客户端收到lease之后很快就过期了，lease机制就发挥不了作用；如果服务器的时间比客户端的时间快，那么就比较危险，因为客户端会在服务器已经开始更新元数据的时候继续使用缓存，工程中，通常将服务器的过期时间设置得比客户端的略大，来解决这个问题。为了保持时间的一致，最好的办法是使用NTP（Network Time Protocol）来保证时钟同步。
Lease机制的本质是颁发者授予的在某一有效期内的承诺，承诺的范围是非常广泛的：比如上面提到的cache；比如做权限控制，例如当需要做并发控制时，同一时刻只给某一个节点颁发lease，只有持有lease的节点才可以修改数据；比如身份验证，例如在primary-secondary架构中，给节点颁发lease，只有持有lease的节点才具有primary身份；比如节点的状态监测，例如在primary-secondary架构中监测primary是否正常，这个后文再详细介绍。
工程中，lease机制也有大量的应用：GFS中使用Lease确定Chuck的Primary副本, Lease由Master节点颁发给primary副本，持有Lease的副本成为primary副本。chubby通过paxos协议实现去中心化的选择primary节点，然后Secondary节点向primary节点发送lease，该lease的含义是：“承诺在lease时间内，不选举其他节点成为primary节点”。chubby中，primary节点也会向每个client节点颁发lease。该lease的含义是用来判断client的死活状态，一个client节点只有只有合法的lease，才能与chubby中的primary进行读写操作。
总结
本文主要介绍分布式系统中的分片相关问题，包括三种分布方式：hash、一致性hash、range based，以及各自的优缺点。分片都是按照一定的特征值来进行，特征值应该从应用的使用场景来选取，并结合MongoDB展示了特征值（mongodb中的sharding key）对数据操作的影响。分片信息（即元数据）需要专门的服务器存储，元数据服务器是分布式存储系统的核心，因此需要提到其可用性和可靠性，为了减轻元数据服务器的压力，分布式系统中，会在其他节点缓存元数据，缓存的元数据又带来了一致性的挑战，由此引入了Lease机制。
references
刘杰:《分布式系统原理介绍》
Distributed systems for fun and profit
Hadoop NameNode 高可用 (High Availability) 实现解析
Leases: An Efficient Fault-Tolerant Mechanism for Distributed File Cache Consistency
以上是关于带着问题学习分布式系统之数据分片的主要内容，如果未能解决你的问题，请参考以下文章