mongodb shard 分片集群基础概念

目录

一、shard 集群

二、Config Server

1、config.shards

2、config.database

3、config.collection

4、config.chunks

5、config.settings

6、其他

三、shard机制

1、Primary Shard

2、Shard Key

2.1 范围分片

2.2 哈希分片

2.3 Shard Key重定义

2.4 版本约束

2.5 Shard Key的使用

3、chunk

 4、Balancer

四、集群备份

1、官方方案

2、鼎甲方案

3、爱数方案

五、参考文献 

---

一、shard 集群

    副本集群中，所有节点维护同一份数据。可以承载部分读请求压力，解决高可用的问题。但随着业务场景的增长，会出现下面的问题：

        1、数据容量超出单机磁盘容量

        2、活跃的数据集超出节点内存容量，需要从磁盘读取，使得读性能下降

        3、写入量超出primary节点主机的IOPS（Input/Output Per Second）容量

    当副本集群性能受限时，存在两种解决方案：

        1、垂直扩容：提高副本集中单节点的性能，提高 CPU、内存、带宽

        2、水平扩容：将任务分片，分给多个副本集群。

    shard 集群就是属于水平扩容方案，下面是 shard 集群的基本架构（插图选自华为 DDS 官网）：

        1、Mongos（Router）：作为 shard 集群的入口，对用户请求进行路由、分发与合并（操作结果）。可通过部署多个mongos实现高可用。

        2、Config Servers ： 存储集群元数据和集群配置；可通过部署副本集实现高可用。

        3、Shard ：存储分片后的用户数据，不同的Shard存储不同的数据。

    应用程序通过 JDBC 连接 mongos 节点实现与整个集群交互。Mongos 则会根据客户端的请求来向后端不同的 Shard 进行请求的发起。

    如下图所示，若对 Collection1 进行读写，Mongos 会和 Shard A 和 Shard B 进行请求交互，如果读写 Collection2，那么 Mongos 只会和 Shard A 进行数据交互。

    可以通过拼接合理的 ConnectionStringURL 连接 shard 集群，如果使用单个 mongos 进行连接，会有风险。

# mongodb://  前缀，代表这是一个  ConnectionString URI 连接地址。
# username:password@ 连接 MongoDB 实例的用户名和密码，使用英文冒号（:）分隔。
# hostX:portX     实例的连接地址和端口号。
# /database       鉴权数据库名，即数据库账号所属的数据库。
# ?options        指定额外的连接选项。
mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]
# 举例：用户名为 user，密码为 password，然后来连接 mongos1 和 mongos2，它们的端口都是 3717，鉴权数据库是 admin
mongodb://user:password@mongos1:3717,mongos2:3717/admin

二、Config Server

    Config server存储Sharded cluster的所有元数据，所有的元数据都存储在config数据库。

    了解该节点内集合的作用很有必要。

mongos> use config
switched to db config
mongos> db.getCollectionNames()
["shards","actionlog","chunks","mongos","collections","lockpings","settings","version","locks","databases","tags","changelog"
]

1、config.shards

    config.shards集合存储各个Shard的信息，可通过addShard、removeShard命令来动态的从Sharded cluster⾥增加或移除shard。如下所示，cluster⽬前拥有2个shard，均为复制集。如下所示，cluster⽬前拥有2个shard，均为复制集。

mongos> db.addShard("mongo-9003/10.1.72.135:9003,10.1.72.136:9003,10.1.72.137:9003")
mongos> db.addShard("mongo-9003/10.1.72.135:9003,10.1.72.136:9003,10.1.72.137:9003")
mongos> db.shards.find()
{ "_id" : "mongo-9003", "host" : "mongo-9003/10.1.72.135:9003,10.1.72.136:9003,10.1.72.137:9003" }
{ "_id" : "mongo-9004", "host" : "mongo-9004/10.1.72.135:9004,10.1.72.136:9004,10.1.72.137:9004" }

2、config.database

    config.databases集合存储所有数据库的信息，包括DB是否开启分⽚，primary shard信息，对于数据库内没有开启分片的数据库/集合，所有的数据都会存储在数据库的primary shard上。

    如下所示：数据库 shtest 开启了分片，primary shard为 mongo-9003；test 没有开启分片，则默认存储在 primary shard上。

mongos> sh.enableSharding("shtest")
{ "ok" : 1 } 
mongos> db.databases.find() 
{ "_id" : "shtest", "primary" : "mongo-9003", "partitioned" : true } 
{ "_id" : "test","primary" : "mongo-9003", "partitioned" : false }

3、config.collection

    数据分片是针对集合维度的，某个数据库开启分片功能后，如果需要让其中的集合分⽚存储，则需调用 shardCollection 命令来针对集合开启分片。

    如下命令，针对shtest数据里的hello集合开启分片，使⽤x字段作为shard key来进行范围分片。

mongos> sh.shardCollection("shtest.coll", {x: 1})
{ "collectionsharded" : "shtest.coll", "ok" : 1 }mongos> db.collections.find()
{ "_id" : "shtest.coll", "lastmodEpoch" : ObjectId("57175142c34046c3b556d302"), "lastmod" : ISODate("1970-02-19T17:02:47.296Z"), "dropped" : false, "key" : { "x" : 1 }, "unique" : false }

4、config.chunks

    集合分片开启后，默认会创建⼀个新的chunk，shard key取值[minKey, maxKey]内的⽂档（即所有的⽂档）都会存储到这个 chunk。当使⽤ Hash 分⽚策略时，可以预先创建多个chunk，以减少chunk的迁移。

mongos> db.chunks.find({ns: "shtest.coll"})
{ "_id" : "shtest.coll-x_MinKey", "ns" : "shtest.coll", "min" : { "x" : { "$minKey" : 1 } }, "max" : { "x" : { "$maxKey" :1 } }, "shard" : "mongo-9003", "lastmod" : Timestamp(1, 0), "lastmodEpoch" : ObjectId("5717530fc34046c3b556d361") }

    当chunk⾥写⼊的数据量增加到⼀定阈值时，会触发chunk分裂，将⼀个chunk的范围分裂为多个chunk，当各个shard上chunk数量不均衡时，会触发chunk在shard间的迁移。如下所示，shtest.coll的⼀个chunk，在写⼊数据后分裂成3个chunk。

mongos> use shtest
mongos> for (var i = 0; i < 10000; i++) { db.coll.insert( {x: i} ); }
mongos> use config
mongos> db.chunks.find({ns: "shtest.coll"})
{ "_id" : "shtest.coll-x_MinKey", "lastmod" : Timestamp(5, 1), "lastmodEpoch" : ObjectId("5703a512a7f97d0799416e2b"), "ns" : "shtest.coll", "min" : { "x" : { "$minKey" : 1 } }, "max" : { "x" : 1 }, "shard" : "mongo-9003" }
{ "_id" : "shtest.coll-x_1.0", "lastmod" : Timestamp(4, 0), "lastmodEpoch" : ObjectId("5703a512a7f97d0799416e2b"), "ns" :"shtest.coll", "min" : { "x" : 1 }, "max" : { "x" : 31 }, "shard" : "mongo-9003" }
{ "_id" : "shtest.coll-x_31.0", "lastmod" : Timestamp(5, 0), "lastmodEpoch" : ObjectId("5703a512a7f97d0799416e2b"), "ns": "shtest.coll", "min" : { "x" : 31 }, "max" : { "x" : { "$maxKey" : 1 } }, "shard" : "mongo-9004" }

5、config.settings

    config.settings 集合⾥主要存储 sharded cluster 的配置信息，⽐如 chunk size，是否开启 balancer 等

mongos> db.settings.find()
{ "_id" : "chunksize", "value" : NumberLong(64) }
{ "_id" : "balancer", "stopped" : false }

6、其他

    config.tags 主要存储 sharding cluster 标签（tag）相关的信息，以实现根据 tag 来分布 chunk 的功能

    config.changelog 主要存储 sharding cluster ⾥的所有变更操作，⽐如 balancer 迁移 chunk 的动作就会记录到 changelog ⾥。

    config.mongos 存储当前集群所有 mongos 的信息

    config.locks 存储锁相关的信息，对某个集合进⾏操作时，⽐如 moveChunk，需要先获取锁，避免多个 mongos 同时迁移同⼀个集合的 chunk

三、shard机制

1、Primary Shard

    新建的数据库默认是未分片的，先存储在一个 Shard 上面，该 Shard 称为 Primary Shard。

    当创建一个新的 database 时，系统会根据各个 shard 目前存储的数据量，选择一个数据量最小的 shard 作为新 database 的 primary shard。如下图，选择了 Shard A作为Database 3的Primary Shard。

    Database 建立并确定 Priamry Shard后将进行分片操作。Shard集群的分片支持集合的级别。已经被分片的集合被切分为多份保存在 shard 上。

sh.enableSharding("<database>")
sh.shardCollection("<database>.<collection>", {<key> : <direction>, ... } )
# <key> : 分片键字段的名字
# <direction> : {1 | -1 |"hashed"} 。  1 | -1 : 基于范围分片键，"hashed" : 哈希分片键

2、Shard Key

    Shard key 是分片时依据的数据库字段。Shard Key 必须是一个索引。非空集合须在 shardCollection 前创建索引；空集合 shardCollection 自动创建索引。

    确定了 Shard Key 后，如何进行分片呢？主要分为两种：范围分片与哈希分片。这两种方法都存在优缺点，依据实际情况使用。

2.1 范围分片

    依据 Shard Key的范围进行分片，每个 Shard 中存放一段 Shard Key 范围的数据。

    如下图所示，是一个基于 x 的范围分片，数据被分为了 4 部分，切割点分别是 x:-75、x:25、x:175 值相近的数据是相邻的。

    这种情况下，可以很好的满足范围查询的需求。但是如果是基于分片键的单调写入，由于数据都会由于所有的写入都会被最后一个 Chunk 来承载，所以这样就无法很好的扩充写能力。

 

2.2 哈希分片

    根据 ShardKey 计算哈希值，基于哈希值进行数据分片。

    如下图所示。根据 x 计算出的哈希值分类存放在 Shard中。这样无论是否单调写入都可以扩充写能力，但无法实现 x 字段的范围查找。

    需要注意，哈希分片只支持单个字段的哈希分片。   

    4.4 以后的版本，可以将单个字段的哈希分片和一个到多个的范围分片键字段来进行组合。比如{x:1, y:"hashed"}，指定 x 范围分片，y 是哈希分片方式来进行组合。

2.3 Shard Key重定义

    4.4 版本新增命令，通过分片键增加后缀字段的方式来修改分片键：

db.adminCommand( {refineCollectionShardKey: "<database>.<collection>",key: { <existing key specification>, <suffix1>:<1|"hashed">, ... }
} )
# <existing key specification> : 当前的分片键，新的分片键必须以当前分片键为前缀；
# <suffix1> : 新增的后缀分片键字段；

（1）新的 ShardKey 对应的索引在 RefineCollection-ShardKey 执行前须已经创建完成；
（2）RefineCollectionShardKey 只会修改 Config 节点上的元数据，不会有任何数据迁移，数据的打散随后续正常分裂&迁移而完成；
（3）4.4 版本中支持了 ShardKey 缺失的情况（当做 Null 处理），为了应对并不是所有文档都存在新的 ShardKey 的所有字段；
（4）4.4 版本中支持复合哈希分片键，而在之前的版本中只能支持单字段的哈希分片键。

 

2.4 版本约束

4.4 版本之前：

        ShardKey 大小不能超过 512 Bytes；

        仅支持单字段的哈希分片键；

        Document 中必须包含 ShardKey；

        ShardKey 包含的 Field 不可以修改。

4.4 版本之后:

        ShardKey 大小无限制；

        支持复合哈希分片键；

        Document 中可以不包含 ShardKey，插入时被当做 Null 处理；

        为 ShardKey 添加后缀 refineCollectionShardKey 命令，可以修改 ShardKey 包含的 Field；

4.2 版本之前，ShardKey 对应的值不可以修改；

4.2 版本之后，如果 ShardKey 为非_ID 字段，那么可以修改 ShardKey 对应的值。

2.5 Shard Key的使用

    Mongos 是如何基于请求当中的 shard key 信息来做请求转发，有两种转发行为，一种叫做特定目标的操作（targeted operation），一种叫做广播操作（Broadcast Operations）。

    Targeted Operation ： 根据 shard key 计算出目标 Shard(s)，发起请求并返回结果。

    Broadcast Operations ：将请求发送给所有 Shard，合并查询结果并返回给客户端。

    读请求：若用户请求中携带了 shard key，则直接计算出目标chunk所在 shard； 若用户请求中不包含shard key，则广播用户请求给所有shard，并将结果合并后返回给用户。

    写请求：必须携带 shard key，直接作用于根据 shard key 计算出的目标chunk所在shard。

 

3、chunk

    MongoDB 基于 ShardKey 将 Collection 拆分成多个数据子集，每个子集称为一个 Chunk。shardedCollection 的数据按照 ShardKey 划分为 MinKey ~ MaxKey 区间，每个 Chunk 有自己负责的一个区间（前闭后开）。存储 ShardedCollection 的 Shard 上有该 Collection 的一个或多个 Chunk ；

    如下图所示：分片的集合是基于 x 的范围分片，数据被分成了 4 个 Chunk， Chunk 1 : [minKey, -75) ;Chunk2 : [-75, 25) ; Chunk3 : [25, 175) ; Chunk4 :[175, maxKey)是个前闭后开的区间。ShardA 是持有 Chunk1 和 Chunk2，而 ShardB 和 ShardC 则分别持有 Chunk3 和 Chunk4。

    chunk有默认的大小限制（64MB，可配置chunk size），超出指定大小后会自动分裂（chunk splits），用户也可以手动进行分裂。

    为了保证数据负载均衡，MongoDB 支持 Chunk在 Shard 间迁移，称为 Chunk Migration。Chunk迁移可自动触发，也可以手动触发。当 Chunk 在 Shard 之间分布不均时，Balancer 进程会自动触发。

    Chunk 迁移的影响：影响 Shard 使用磁盘的大小；增加 网络带宽 及 系统负载，这些会对系统性能造成影响。

    Chunk 迁移的约束：每个 Shard 同一时间只能有一个 Chunk 在进行迁移；不会迁移 Chunk 中文档数量是平均 Chunk 文档数 1.3 倍的 Chunk // 4.4 提供选项支持。

 4、Balancer

    Balancer 是 MongoDB 的一个后台进程，用保证集合的 Chunk 在各个 Shard 上是均衡的。

    Balancer 运行在 ConfigServer 的 Primary 节点。 默认为 开启状态。

    当分片集群中发生 Chunk 不均衡的情况时，Balancer 将触发 Chunk 从 Chunk 数量最多的 Shard 向 Chunk 数量最少的 Shard 上迁移。每个 Shard 同一时间只能有一个 Chunk 在进行迁移；

    如图所示：Chunk 的数量小于 20，迁移阈值是 2，随着 Chunk 数量增大，迁移阈值分别增长为 4 和 8。

四、集群备份

1、官方方案

        自部署的集群，官方提供了两种方案，这两种方案大体流程没啥区别，主要区别在于对数据库的备份手段，例如对文件系统进行快照或使用 mongodump。单个人觉得，mongodump 工具属于逻辑备份，性能底下，对于备份期间需要锁住数据库的情况，客户应该接受不了。不过个人没有实践过，无法给出具体的性能参数。下面介绍通用流程。

（1）寻找合适的备份窗口

        数据块迁移、重新分片和模式迁移操作会导致备份不一致。

（2）停止负载均衡器

        为了防止数据块迁移破坏备份，使用 sh.stopBalancer() 方法停止平衡器.

        如果当前正在进行均衡操作，则停止操作会等待均衡操作完成后再继续执行。

        要验证负载均衡器是否已停止，请使用 sh.getBalancerState() 方法：

use config
sh.stopBalancer()
while( sh.isBalancerRunning().mode != "off" ) {print("waiting...");sleep(1000);
}

 （3）锁定集群

    写入数据库可能会导致备份不一致。锁定分片集群以防止数据库被写入。

    若要锁定分片集群，请使用 db.fsyncLock() 方法：

db.getSiblingDB("admin").fsyncLock()

    在配置服务器的 mongos 和主 mongod 上运行以下聚合管道。要确认锁定，请确保  fsyncLocked 字段返回 true ，fsyncUnlocked 字段返回 false。

（4）备份 Config Server 的主节点

（5）备份每个 Shard 主节点

（6）解锁集群

    备份完成后，您必须解锁群集以允许恢复写入。

    解锁集群，在配置服务器的 mongos 和主 mongod 上运行以下聚合管道。要确认解锁，请确保 fsyncLocked 字段返回 false ， fsyncUnlocked 字段返回 true。

db.getSibling("admin").fsyncUnlock()

（7）重启负载均衡器

    重启负载均衡器执行下面的语句。

sh.startBalancer()

2、鼎甲方案

        基于官网，内容大同小异。

        DBackup 新版基于快照技术，实现高效的流式备份恢复方案：

        （1）自动选择从节点备份，避免影响主节点业务；

        （2）仅恢复主节点，从节点将自主同步；

        （3）支持跨集群恢复，不受限于原始架构；

3、爱数方案

        基于官网，内容大同小异。

五、参考文献 

 《mongodb入门实战》阿里云

mongodb.com/zh-cn/docs/manual/https://www.mongodb.com/zh-cn/docs/manual/