MongoDB的架构

前言

最近尝试用MongoDB替换Mysql，由于Mysql的写能力限制， 32C128G的实例同时写入1M的数据CPU飙升，更不用说如果是并发的情况。MongoDB通过sharding能够很好地解决写能力扩展的问题, 故作一篇网上大神博客的小结. MongoDB的架构大致如下图所示(v3.2):

与RBDMS的主要区别

MongoDB和RDBMS的主要区别在于：

RBDMS的数据记录是平的，而MongoDB的数据单元Document是可以嵌入的，比如一个CSV文件，一个字段一个值; 但是mongo可以一个字段可以存储多个值如:{'key': [v1,v2]}, 通过不同组合还可以生成更复杂的数据结构。
RBDMS所有的数据结构schema必须预定义， MongoDB不需要预定义, Document可以存储任何结构的数据。
MongoDB是没有join查询操作的。 RBDMS数据库设计的核心讲究Normalization, 数据越结构化，冗余越少越好。而MongoDB鼓励denormalized, 通过数据冗余做到join查询。
MongoDB的数据一致性需要客户端来维护， mongo没有ACID的Isolation概念，一个并发客户端可能读到另一个并发用户修改的数据。
同样地, transaction事务也是不存在，（注意， 4.0这个版本支持ACID的事务）。MongoDB原子性操作只能做到document级别。
而正是移除关系型数据库部分特性， mongo才能做到更好的扩展性和轻量级，这两条正适合用于处理大数据。

查询处理

Mongo的Collection可以认为是关系型数据库的table， Document可以认为是关系型数据库的records. 并且不需要预创建数据库和collection, 如下是一些基本操作,

In [1]: import pymongo

In [2]: p = {'first':"Dave", 'lastname': "He"}

In [3]: client = pymongo.MongoClient()

In [4]: db = client.test_database

In [6]: db.person.insert_one({'first':'NN', 'lastname':'HE'})
Out[6]: <pymongo.results.InsertOneResult at 0x21f9710>

In [7]: db.person.find_one()
Out[7]: 
{u'_id': ObjectId('5ba858bc91046605e83667dd'),
 u'first': u'Dave',
 u'lastname': u'He'}

可以创建索引加速查询, MongoDB的索引以Btree的数据结构存储, 所以支持范围查询. 由于document本身就是一棵树, 索引可以嵌套到document下层里的某个值. 也可以创建符合索引, 如db.person.ensureIndex({lastname:1, firstname:1}). 索引也可以是单键多值array.

In [5]: db.person.index_information()
Out[5]: {u'_id_': {u'key': [(u'_id', 1)], u'ns': u'test_database.person', u'v': 2}}

mongo默认给_id创建索引, 也可以给一个字段建新的索引, 创建索引可以是前台线下模式或者后台在线模式, 如果是线下模式, 如果是多副本集,需要考虑这些副本索引做到滚动更新.

In [6]: r = db.person.create_index([('lastname', pymongo.ASCENDING)], unique=True)

In [8]: db.person.index_information()
Out[8]: 
{u'_id_': {u'key': [(u'_id', 1)], u'ns': u'test_database.person', u'v': 2},
 u'lastname_1': {u'key': [(u'lastname', 1)],
  u'ns': u'test_database.person',
  u'unique': True,
  u'v': 2}}

但开始查询时如果有多个查询条件, mongo总是先尝试使用单个最佳索引找到符合的数据集, 然后根据后续的其它条件迭代查询.
那么如果一个collection的多索引是如何配合来加速查询的呢?
当一个查询执行时, mongo会给每个索引创建一个执行计划, 每个索引轮流执行查询, 直到所有索引执行完查询, mongo记录下最快执行查询的索引,后续就会使用这个索引查询, 直到后续一定量的数据更新, 才会重新执行上述流程.
针对执行计划每次查询只用一个索引的特点, 查询时注意查询条件, 并且创建主键以外的复合索引来加速查询至关重要! 同时也要注意索引的开小, 删除不必要的索引.

存储模型

MongoDB通过内存mapping文件把存储在磁盘上数据文件直接映射到内存byte array, 数据的访问逻辑是通过指针算法实现的. 每个collection存在单独的一个namespace文件(记录元信息)和多个可扩展(extent)数据文件.

每个collection的数据被组织在extent文件中, 每个extent是一段连续的磁盘空间,使用双向链表连接extent. extent包含了多个document, 每一个document也会和其相邻的document相连, 实际数据就以Bson的格式存储在上图中的DocRecord. 一个extent会指向所包含的document链表的头和尾(当然extent和extent之间也是连起来的).

任何修改数据会直接在原地进行. 为了预防数据修改后大小超过了原来记录的分配的空间, 整条记录会被移动到一个更大的区域(附带一些填充占位的字节, 这些填充的字节当作缓存空间用于将来可以用于放置更大的空间, 那么具体填充多少字节, 这就需要每个collection有一个统计值, 专门记录修改的统计数据), 原来空出来的空间将会被释放, 有一张表会记录free list的大小.
基于上述移动空间的设计, 我们可以预见数据会变得片段化, 所以mongo需要周期性地运行compact命令, 会把数据移动到连续的空间内, 以提高IO性能.这个操作通常需要在线下进行, 副本集则需要滚动更新.

索引是通过Btree实现的, 每个Btree的节点含有一个key数字和指向左边子节点的指针.

数据的更新和事务

更新一条记录是在原数据上直接做修改:

In [6]: db.person.update_one({'first':'Dave'}, {"$set":{"first":'David'}})

In [8]: db.person.find_one()
Out[8]: 
{u'_id': ObjectId('5ba858bc91046605e83667dd'),
 u'first': u'David',
 u'lastname': u'He'}

mongo写的时候可以指定多种方式(policy), 以表示写是否成功了, 如数据已经在磁盘持久化了; 或者数据已经在多个副本集传递(可以通过getLastError拿到执行返回值). 另外, mongo是可以给副本集打tag, 所以可以基于tag制定写策略.

由于所有从mongo读到的数据都是过去的快照, 读到的数据可能已经被其它客户修改, 所以如果有一致性要求, 每次修改数据前可以先读一次验证数据, 然后再做修改.
mongo可以在修改前验证条件, 使用findAndModifyapi更新数据. 如果是pymongo那么update会帮做验证条件, 不需要额外做这一步.

mongo没有事务的概念(4.0以前), 对每个document的操作都是原子性的, 但是如果多个document被修改, 原子性是不能保证的.所以客户端需要自行实现多文件更新原子性.一个常见的技术是: 首先创建一个独立的文档(叫做transaction), 把所有需要更新的文档连接起来; 然后把所有需要更新的文档反链到transaction; 接着如下图做二段式提交完成事务.

复制模型

Mongdb通过副本集(replica set)做到高可用和读查询的负载均衡(没有做到写负载均衡), 通过把数据复制到多个服务器做到冗余. 具体包括一个主要DB和多个二级DB. 为了数据一致性, 所有修改都会在主DB操作然后异步复制到副本. 副本集之间的节点通过心跳同步状态, 如果有一个节点失去心跳(挂了), 那么就会失去membership. 如果后面恢复了那么这个节点会重新加入集群, 通过和主DB通信,获取changelog恢复到最新的数据. 如果主DB的changelog没有覆盖这个节点挂掉到恢复的全部日志, 那么需要重新导入主DB的所有数据.

如果主DB挂掉了, 那么集群将进行选举, 通过节点优先级,运行时间等条件从二级DB中选出新的主DB, 由于二级DB采用异步复制的形式, 这个新选出来的主DB可能不是最新的数据版本.

客户端驱动库需要实现找到主DB的功能, 在开始连接数据库后, 客户端需要发出isMaster的指令从当前集群中找到主DB和所有副本, 然后客户端会把大部分请求发给主DB, 部分读请求发给二级DB; 这个isMaster的指令会周期性的发送, 以便同步当前集群信息. 所有挂掉的节点, 客户端会强制中断连接.

MongoDB有一个特殊的二级DB叫Delayed Slave, 它是延迟一段时间同步主数据库. 主要用来恢复短期误删的的数据.
在读取查询时, 可以指定从二级DB读取数据, 这样读到的数据可能不是最新的, 但是可以使用这个特性做到查询的负载均衡. 客户端可以ping二级DB, 挑选最快的节点查询数据.

分片模型

想要做到写负载均衡, 需要使用MongoDB的sharding. 在sharding配置中, 一个collection可以通过partition key被分割到多个chunks内(一组key的范围), 多个chunks被分配到不同的shards, 每个shards都是副本集(replica set). 通过sharding, mongo可以存储无限量的数据, 这点在大数据应用场景十分重要.

在sharding模式下, 客户端需要连接到MongoS, 它是一个路由服务器把客户端的请求发送到合适的分片. 对于insert/delete/update修改操作请求, 如果包含partition key, 那么基于chunk/shard映射表信息(从config server得到,并且本地缓存), 路由服务器可以找到对应chunk的节点服务器. 如果是读查询, 路由服务器会检测partition key是否包含在查询的部分条件中, 如果是那么就可以找到对应的主或二级shard; 但是也有可能查询条件不含有partition key, 那么路由会把请求发给每个shard.如果查询需要排序, 同时partition key在排序的条件中, 那么路由会一次按partition key排序shards; 如果不包含partition key, 路由服务器会把请发给每个shard, 然后做merge-sort.

另一方面, 路由服务器需要保障每个shards中的数据chunks大致一样多. 当不平衡的条件被检测到, 路由器会通知chunk较多的shard触发迁移工作. 这个迁移工作是线上进行的, 数据迁移时进行多次delta检查, 直到最后数据迁移完毕, 目标服务器会通知config server新的shard信息, 通知源服务发送StaleConfigException给路由服务器, 让路由器重新读取配置信息. 后面的某个时间点, 源服务器旧的shard数据将会被删除. 如果在迁移的过程中发生高频率的更新操作, 那么迁移会停止,路由服务器会选择新的chunk去迁移数据.