通過一步步創建sharded cluster來認識mongodb

　　mongodb是目前使用非常廣泛的nosql(not only sql)之一，在db engines上排名非常靠前，下圖是5月份的排名：

　　

　　可以看到前面四個都是傳統的關系型數據庫，而mongodb在nosql中拔得頭籌。本文會簡單介紹mongodb的一些特性，然后通過在Linux環境下一步步搭建sharded cluster來學習mongodb。本文實驗的mongodb是mongodb3.0，可能與最新的版本（mongodb3.4）在細節之處略有差異。

 

Mongodb特性

 　　官方一句話就能概括Mongodb的特點：

　　MongoDB is an open-source document database that provides high performance, high availability, and automatic scaling.

　　開源、基於文檔（document oriented）、高性能、高可用、自動伸縮。

　　開源：

　　這個好處就不用多說了，GitHub上有源碼。

　　 面向文檔：

　　文檔（document）在很多編程語言都有類似的數據結構，各種table、map、dict，再也不用使用DAO（data access object)。比如在python中，document與dict對應，array與list對應。

      document也支持嵌套的document和array，這樣的話也能部分解決關聯查詢（當然，雖然把相關信息放在一個嵌套的document降低了關聯查詢的開銷，但在某些情況不得不需要關聯查詢的時候還是有點頭疼）

　　由於基於document，所以就schema free（模式自由）啦，使用關系型數據的同學都知道，線上修改表結構是多么麻煩的一件事情。但在mongodb中，增該刪一個字段太容易了，這個也是最后開發人員喜歡的一點，比如游戲服務器，玩家的持久化數據會不停的變化，每次更新都會增加一些功能，也就回增加一些需要持久化的字段，用mongodb就很合適。

　　

　　高性能：

　　支持嵌套的document，在關系型數據庫中需要聯合查詢的耗時操作，在mongodb中一條查詢就能搞定

　　豐富的索引支持，而且索引還支持嵌套文檔和數組

　　在使用了sharding機制的情況下，讀寫操作可以路由到不同的shard，提到了集群的並發性能

 

　　高可用：

　　要想mongodb高可用，那么就得使用mongodb的復制機制：replica set

　　replica set通過異步復制（Asynchronous replication）和自動Failover來保證可用性

　　后面會專門介紹replica set

 

　　自動擴展（水平拆分）：

　　在關系型數據中，當單個表數據量過大的時候，一般會通過垂直分表或者水平分表的方式來提到數據庫吞吐量。在mongodb中，sharding是其核心功能之一，提供了自動的水平擴展（horizontal scalability）來對數據量比較大的集合進行拆分，sharding將同一個集合的不同子集數據放在不同的機器上，當應用程序選擇好適當的sharding key，可以將讀寫操作路由到某一個shard上，大大提高了集群吞吐性能。

　　后面會專門介紹sharding cluster的組成

 

預備知識

　　本文並不涵蓋mongodb的基礎知識，但是為了后面介紹sharding知識，以及搭建sharded cluster，在這里介紹一下 _id這個特殊的字段（field）

　　默認情況下，mongodb會在集合的_id字段上創建unique index。如果沒有在持久化的文檔中包含_id，那么mongodb會自動添加這個字段，其value是一個ObjectId。

　　mongodb官方建議_id使用ObjectId、或者自然唯一標示（unique identifier）、說着自增的數字、或者UUID。_id在sharding相關的CRUD中有特殊性，具體使用的時候可以參加文檔。

　　另外，為了后文描述方便，這里聲明幾個在mongodb中的概念

　　DB：與mysql的DB對應

　　collection、集合： 與mysql的table對應

　　document：與mysql的record對應

replica set

　　MongoDB中通過 replica set（復制集）來提供高可用性：冗余與自動failover。

　　復制集是說將 同一份數據的多分拷貝放在不同的機器（甚至不同的數據中心）來提高容錯。一個典型的replica set由一組mongod實例組成，其中有且僅有一個節點提供寫操作，稱之為primary，primary也是默認的讀節點。同時，replica set中可包含一個到多個secondary節點，secondary節點只提供讀操作。如下圖所示：

　　

 

　　應用程序通過驅動與Primary連接，所有的寫操作都在Primary上進行，同時primary會將這些操作寫到oplog（operation log）中，secondary通過異步復制oplog，然后在本地數據集上執行oplog中的操作，這樣就達到了數據的一致性。從這里可以看到，雖然secondary和primary維護的上同一份數據，但是其變更是要遲於primary的。

　　如果應用程序對數據的實時性要求不太高，比如評論數據、聚合數據等，那么可以從secondary讀取，這樣可以做到讀寫分離和高並發。如果拷貝放在不同的數據中心，能更好的提供數據局部性（data locality）和分布式服務的可用性。

 

　　我們看到，如果一個Secondary不能正常工作了（可能是進程crash、物理損壞、網絡故障），對應用程序來說影響並不大。但是如果primary不能工作了呢？這個時候mongodb的automatic failover就開始發揮作用了。

　　在replica set中的所用mongod節點之間都會有心跳（heartbeat）存在，如果超過一定時間其他節點沒有收到primary的心跳，那么就認為primary掛掉了。 可被選舉的secondary會投票選舉出新的primary。整個過程如下所示：

 

　　自動的failover 雖然保證了mongodb的高可用性，但是在primary到secondary的切換過程中，這一段時間，mongodb是無法提供寫操作的。表現就是對於應用程序的數據庫操作請求會返回一些錯誤，這個時候應用程序需要識別這些錯誤，然后做重試。

　　

　　除了Primary和Secondary，在replica set中還可以存在存在另外一種節點：Arbiter。Arbiter與Secondary節點的區別在於，Arbiter不持久化數據（do not bearing data),  自然也不可能在Primary掛掉的時候被選舉。Arbiter的作用在於投票：為了選出新的primary，secondary投票規則是少數服從多數，如果replica set中的節點數目是偶數，那么就可能出現“平局”的情況，所以加入一個Arbiter就可以以最小的代價解決這個問題。

 

 　　Arbiter不持久化數據，所以占用的磁盤空間也很少，對硬件的要求也不高。官方建議，Arbiter不要和primary或者secondary放在同一個物理主機上。

　　在后面的演示中，也會在replica set中加入一個Arbiter，減少磁盤占用。

sharded cluster

　　所謂sharding就是將同一個集合的不同子集分發存儲到不同的機器（shard）上，Mongodb使用sharding機制來支持超大數據量，將不同的CRUD路由到不同的機器上執行，提到了數據庫的吞吐性能。由此可見，sharding是非常常見的scale out方法。

　　

 

　　如上圖所示，一個集合（Collection1）有1T的數據，原本放在一個單獨的數據庫中，通過sharding，將這個集合的數據放在四個獨立的shard中，每一個shard存儲這個集合256G的數據。每個shard物理上是獨立的數據庫，但邏輯上共同組成一個數據庫。

　　一個sharded cluster由一下三部分組成：config server，shards，router。如圖所示：

　　

 

　　shards：

　　存儲數據，可以是單個的mongod，也可以是replica set。在生產環境中，為了提高高可用性，都會使用replica set。存儲在mongod上的數據以chunk為基本單位，默認的大小為64M，后面會介紹shard上數據的分裂（split）與遷移（migration）

　　config server：

　　存儲集群的元數據（metadata），即數據的哪一部分放在哪一個shard上，router將會利用這些元數據將請求分發到對應的shards上，shards上chunk的遷移也是config server來控制的。

　　router:

　　mongos實例，在一個集群中直接為應用程序提供服務，利用config server上的元數據來制定最佳的查詢計划。

 

　　數據分割（data partition）：

　　從前文知道，MongoDB在collection這個級別進行數據的切塊，稱之為sharding。塊的最小粒度是chunk，其大小（chunkSize）默認為64M。

　　當一個集合的數據量超過chunkSize的時候，就會被拆分成兩個chunk，這個過程稱為splitting。那么按照什么原則將一個chunk上的數據拆分成兩個chunk，這就是Sharding key的作用，Sharding key是被索引的字段，通過sharding key，就可以把數據均分到兩個chunk，每一個document在哪一個chunk上，這就是元數據信息。元數據信息存放在config server上，方便router使用。

　　如果sharding cluster中有多個shard，那么不同shard上的chunk數目可能是不一致的，這個時候會有一個后台進程（balancer）來遷移（migrate）chunk，從chunk數目最多的shard遷移到chunk數目最少的chunk，直到達到均衡的狀態。遷移的過程對應用程序來說是透明的。

　　如下圖所示，遷移之前ShardA ShardB上都有3個chunk，而Shard C上只有一個Chunk。通過從ShardB上遷移一個chunk到ShardC，就達到了一個均衡的狀態。　　

　　splitting和migration 的目的是為了讓數據在shards之間均勻分布，其根本目標是為了將對數據的CRUD操作均衡地分發到各個shard，提高集群的並發性能。

　　

Sharded cluster搭建

　　聲明，本章節只是演示Sharded Cluster的搭建過程，與生產環境還是有較大差異，不過我也會在文中盡量指出這些差異。首先需要注意的是，本文的演示不涉及到鑒權(--auth)，但在生產環境中鑒權是非常重要的，相信大家都還記得春節期間Mongodb被劫持、被攻擊的事件。

　　前文已經提到，一個典型的Sharded Cluster包括router(mongos)、config server和shards，其中每個shard都可以是單點(standalone)或者復制集(replica set）。 接下來的演示包括一個router， 三個config server，兩個shard。每一個shard都是有一個primary、一個secondary和一個arbiter組成的replica set。

　　在開始之前，首先預定義好所有需要用到的變量，如下所示：

#!/bin/bash
export BIN_HOME=/usr/local/mongodb/bin
export DB_PATH=/home/mongo_db/data
export LOG_PATH=/home/mongo_db/log

LOCAL=127.0.0.1

#config rs
export RS1_1_DB_PATH=$DB_PATH/rs1_1
export RS1_2_DB_PATH=$DB_PATH/rs1_2
export RS1_3_DB_PATH=$DB_PATH/rs1_3
export RS2_1_DB_PATH=$DB_PATH/rs2_1
export RS2_2_DB_PATH=$DB_PATH/rs2_2
export RS2_3_DB_PATH=$DB_PATH/rs2_3

export RS1_1_DB_LOG=$LOG_PATH/rs1_1.log
export RS1_2_DB_LOG=$LOG_PATH/rs1_2.log
export RS1_3_DB_LOG=$LOG_PATH/rs1_3.log
export RS2_1_DB_LOG=$LOG_PATH/rs2_1.log
export RS2_2_DB_LOG=$LOG_PATH/rs2_2.log
export RS2_3_DB_LOG=$LOG_PATH/rs2_3.log

export RS1_1_PORT=27018
export RS1_2_PORT=27019
export RS1_3_PORT=27020
export RS2_1_PORT=27021
export RS2_2_PORT=27022
export RS2_3_PORT=27023

export RS1=rs1
export RS2=rs2

#config config_server
export CONF1_DB_PATH=$DB_PATH/db_conf1
export CONF2_DB_PATH=$DB_PATH/db_conf2
export CONF3_DB_PATH=$DB_PATH/db_conf3

export CONF1_DB_LOG=$LOG_PATH/conf1.log
export CONF2_DB_LOG=$LOG_PATH/conf2.log
export CONF3_DB_LOG=$LOG_PATH/conf3.log

export CONF1_PORT=40000
export CONF2_PORT=40001
export CONF3_PORT=40002

export CONF1_HOST=$LOCAL:$CONF1_PORT
export CONF2_HOST=$LOCAL:$CONF2_PORT
export CONF3_HOST=$LOCAL:$CONF3_PORT

#config route_server
export ROUTE_DB_LOG=$LOG_PATH/route.log

export ROUTE_PORT=27017

 　　可以在會話窗口中將這些命令執行一遍，不過更好的方式是將其保存在一個文件中（如mongodb_define.sh），然后執行這個文件就行了：source mongodb_define.sh

啟動shards（replica set）

　　在這一部分，會創建兩個replica set，分別是rs1， rs2。每個replica set包含三個節點，且其中一個是arbiter。由於兩個replica set創建過程沒什么區別，因此以rs1為例。關於replica set的搭建，可參見mongodb doc中 deploy-replica-set-for-testing部分，講得比較清楚。

　　 step1: 首先得創建好存放數據的目錄：

　　mkdir -p $RS1_1_DB_PATH

　　mkdir -p $RS1_2_DB_PATH

　　mkdir -p $RS1_3_DB_PATH

　　PS: -p means "no error if existing, make parent directories as needed"

 

　　 step2: 啟動組成rs1的三個mongod　

　　$BIN_HOME/mongod --port $RS1_1_PORT --dbpath $RS1_1_DB_PATH --fork --logpath $RS1_1_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_2_PORT --dbpath $RS1_2_DB_PATH --fork --logpath $RS1_2_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_3_PORT --dbpath $RS1_3_DB_PATH --fork --logpath $RS1_3_DB_LOG --replSet $RS1 --smallfiles --nojournal

　　關於mongod的啟動選項，可以通過mongod --help查看，在上面的命令行中，--replSet 指定了replica set的名字， --smallfiles 聲明使用更小的默認文件， --nojournal表明不開啟journaling機制。注意，在這個地方不開啟journaling是因為實驗環境磁盤空間有限，而所有的mongod實例都在這個機器上，在生成環境中，一定要開始journaling，這個是mongodb durability的保證。

　　

　　 step3：初始化復制集rs1

　　在這一步，需要通過mongdb的客戶端mongo連接到復制集的任何一個節點，對復制集初始化，這里連接到RS1_1（端口為27018）：

　　mongo --port $RS1_1_PORT

　　先來看一下現在復制集的狀態（PS：下面所有以 > 開頭的命令行都表示是在mongo這個交互式客戶端輸入的指令）

　　> rs.status()

　　

　　可以看到這個復制集還沒有初始化

　　>config = {

　　　　_id : "rs1",
　　　　members : [
　　　　　　{_id : 0, host : "127.0.0.1:27018"},
　　　　　　{_id : 1, host : "127.0.0.1:27019"},
　　　　　　{_id : 2, host : "127.0.0.1:27020", arbiterOnly: true},
　　　　]
　　}

　　>rs.initiate(config)

　　

　　從config和運行后的復制集狀態都可以看到，RS1_3（127.0.0.1:27020）這個mongod為一個Arbiter，即只參與投票，不持久化數據。另外RS1_1為Primary， RS1_2為Secondary。

　　到此為止，復制集rs1就啟動好了。

 

 

　　關於s2的啟動，下面也給出所有命令。方便讀者實踐

　　mkdir -p $RS2_1_DB_PATH
　　mkdir -p $RS2_2_DB_PATH
　　mkdir -p $RS2_3_DB_PATH

　　$BIN_HOME/mongod --port $RS1_1_PORT --dbpath $RS1_1_DB_PATH --fork --logpath $RS1_1_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_2_PORT --dbpath $RS1_2_DB_PATH --fork --logpath $RS1_2_DB_LOG --replSet $RS1 --smallfiles --nojournal
　　$BIN_HOME/mongod --port $RS1_3_PORT --dbpath $RS1_3_DB_PATH --fork --logpath $RS1_3_DB_LOG --replSet $RS1 --smallfiles --nojournal

　　mongo --port $RS2_1_PORT
　　>config = {
　　_id : "rs2",
　　members : [
　　{_id : 0, host : "127.0.0.1:27021"},
　　{_id : 1, host : "127.0.0.1:27022"},
　　{_id : 2, host : "127.0.0.1:27023", arbiterOnly: true},
　　]
　　}
　　>rs.initiate(config)

啟動config servers

　　mongodb官方建議config server需要三個mongod實例組成，每一個mongod最好部署在不同的物理機器上。這個三個mongod並不是復制集的關系，

　　step1：創建db目錄

　　mkdir -p $CONF1_DB_PATH
　　mkdir -p $CONF2_DB_PATH
　　mkdir -p $CONF3_DB_PATH

　　step2：啟動三個mongod實例：

　　$BIN_HOME/mongod --port $CONF1_PORT --dbpath $CONF1_DB_PATH --fork --logpath $CONF1_DB_LOG --configsvr --smallfiles --nojournal
　　$BIN_HOME/mongod --port $CONF2_PORT --dbpath $CONF2_DB_PATH --fork --logpath $CONF2_DB_LOG --configsvr --smallfiles --nojournal
　　$BIN_HOME/mongod --port $CONF3_PORT --dbpath $CONF3_DB_PATH --fork --logpath $CONF3_DB_LOG --configsvr --smallfiles --nojournal

　　同樣啟動參數中nojournal只是為了節省存儲空間，在生產環境中一定要使用journaling。與創建replica set時mongod的啟動不同的是，這里有一個configsvr 選項，表明這些節點都是作為config server存在。

　　再啟動這三個mongod之后，不會有類似replica set那樣講三個mongod綁定之類的操作，也說明了config server之間是相互獨立的

 

啟動router

　　在Sharded Cluster中，router（mongos）是應用程序連接的對象，一切對mongodb的操作都通過router來路由

　　 step1：啟動mongos

 

　　 $BIN_HOME/mongos --port $ROUTE_PORT --configdb $CONF1_HOST,$CONF2_HOST,$CONF3_HOST --fork --logpath $ROUTE_DB_LOG --chunkSize 32

　　注意這里的可執行程序是mongos，而不是之前的mongod，關於參數，也是可以通過mongos --help查看的。在上面的命令中，--configdb選項指定了三個config server，--chunkSize指定了chunk的大小，單位為M。關於chunksize，默認是64M，雖然可以在初次啟動的時候指定chunksize，但mongodb官方推薦按照 以下方式修改。 在本文中將chunkSize改小的目的，是為了以后實驗的時候更方便觀察數據的拆分和遷移。

　　chunkSize事實上會持久化到config.setting中，連接到mongos可查看：

　　mongo --port $ROUTE_PORT 

　　

　　在上面截圖藍色框中可以看出，現在還沒有任何shard的信息，原因是到現在為止，config servers與replica set還沒有任何關系

 

　　 step2：將在前面創建的兩個replica set（rs1 rs2）加入到Sharded Cluster中

　　 mongo --port $ROUTE_PORT 

　　mongos> sh.addShard('rs1/127.0.0.1:27018')
　　mongos> sh.addShard('rs2/127.0.0.1:27021')

　　PS：為什么需要在rs1后面指定一個mongod的ip port，這個是用來找到對應的mongod，繼而找到相應rs

　　再次查看結果：

　　

　　可以看到已經添加了兩個shard，每一個都是一個replica set。有意思的是Arbiter（比如RS1_3）並沒有顯示在查詢結果中，可能的原因是Arbiter並不持久化數據，顯示在這里也沒有什么意義。

 

　　到此為止，整個Sharded Cluster就算搭建好了，但是還未進入真正使用階段，要發揮Sharded Cluster的作用，我們得指定哪些collection可以被sharding，以及如何sharding

創建sharding key

　　為了演示，我們假設添加一個db叫test_db, 其中有兩個collection，一個是需要sharding的，叫sharded_col；另一個暫時不用sharding，叫non_sharded_col， 當然之后也可以增加新的集合，或者把原來沒有sharding的集合改成sharding。

　　一下操作都需要登錄到router進行： mongo --port $ROUTE_PORT 

　　step1：首先得告知mongodb test_db這個數據庫支持sharding

　　mongos> sh.enableSharding("test_db")
　　{ "ok" : 1 }

　　這個時候可以查看數據庫的狀態，注意，是在config這個db下面的databases集合

　　mongos> use config

　　mongos> db.databases.find()

　　{ "_id" : "admin", "partitioned" : false, "primary" : "config" }

　　{ "_id" : "test_db", "partitioned" : true,  "primary" : "rs1" }

　　從查詢結果可以看到，test_db是支持sharding的（"partitioned" : true）。另外上面加粗部分primary： rs1，這個primary與replica set里面的primary不是一回事，這里的primary是primary shard的意思，我們知道即使在sharded cluster環境中，一些數據量比較小的db也是無需分片的，這些db默認就存放在primary shard上面

 

　　 step2：為需要的collection（即這里的sharded_col）指定sharding key

　　前面已經提到了sharding key的作用，關於sharding key的選擇，是一個比較復雜的問題，sharding key對索引，對CRUD語句的操作都有諸多限制，這一部分以后再細講，在這里默認使用_id做sharding key（_id是mongodb默認為每個集合增加的索引）

　　mongos> sh.shardCollection('test_db.sharded_col', {'_id': 1})

　　接下來看看整個sharded cluster的狀態：

　　

　　sh.status()反應的內容事實上也是來自config整個數據庫的內容，只不過做了一定程度的整合。從上面可以看到，有兩個shard，rs1, rs2;test_db允許sharding，test_db.sharded_col整個collection的sharding key為{"_id": 1}，且目前只有一個chunk在rs1整個shard上。

總結：

　　到目前為止，我們已經搭建了一個有三個config server，兩個shard的sharded cluster，其中每一個shard包含三個節點的replica set，且都包含一個Arbiter。我們可以查看一下剛創建好之后各個mongod實例持久化的數據大小：

　　

　　可以看到，兩個Arbiter（rs1_3, rs2_3）所占的空間要小得多。

　　對於應用程序來說，集群（sharded cluster）和單點（standalone）是有一定差異的，如果需要發揮sharded cluster高性能、高可用的特點，需要根據應用場景精心選擇好sharding key，而sharding key的選擇跟索引的建立以及CRUD語句息息相關，這一部分以后再聊。對於目前搭建的這個實例，簡單測試的話，往sharded_col插入足夠多條document就能看到chunks的拆分和遷移。

 

references：

db engines

the-id-field

replication-introduction

deploy-replica-set-for-testing

deploy-shard-cluster