In-depth understanding of MongoDB sharding management

Preface

In MongoDB (version 3.2.9In the sharded cluster (sharded cluster), a method of horizontally scaling the performance of a database system is to distribute the dataset across different shards (shards), where each shard only saves a part of the dataset. MongoDB guarantees that there will be no duplicate data between different shards, and the sum of the data saved by all shards is the complete dataset. The sharded cluster distributes the dataset, which can distribute the load across multiple shards, with each shard only responsible for reading and writing a part of the data, fully utilizing the system resources of each shard and improving the throughput of the database system.

The dataset is divided into data blocks (chunks), each containing multiple docs, and the data blocks are distributedly stored in the sharded cluster. MongoDB is responsible for tracking the distribution information of data blocks on shards, which data blocks each shard stores, called shard metadata, and stored in the database config on the config server, generally using3There is a config server, and all config databases in the config server must be completely the same. Through mongos, you can directly access the database config to view the metadata of the sharded cluster; the mongo shell provides the sh auxiliary function, which can safely view the metadata information of the sharded cluster.

Querying any shard will only retrieve the data subset of the collection on the current shard, not the entire dataset. The application only needs to connect to mongos and perform read/write operations on it, and mongos will automatically route read/write requests to the corresponding shard. MongoDB makes the underlying shard implementation transparent to the application through mongos, making the application believe that it is accessing the entire dataset.

First, the primary shard

In a sharded cluster, not every collection is distributedly stored. Only after the collection is explicitly sharded using sh.shardCollection(), will the collection be distributedly stored in different shards. For non-sharded collections (un-sharded collection), and its data is only stored in the primary shard (Primary shard). By default, the primary shard refers to the shard that was initially created when the database was created, used to store the data of non-sharded collections in the database. Each database has a primary shard.

Each database in a sharded cluster has a primary shard that holds all the un-sharded collections for that database. Each database has its own primary shard.

For example, a sharded cluster has three shards: shard1, shard2, shard3, in the sharded shard1If the database blog is sharded, MongoDB will automatically create shards2, shard3Create a database blog with the same structure. The primary shard of the database blog is Shard1.

Illustration, Collection2The primary shard is ShardA.

The movePrimary command is used to change the default primary shard of the database. Non-sharded collections will be moved from the current shard to the new primary shard.

db.runCommand({ movePrimary: "test", to: "shard000"1})

After using the movePrimary command to change the primary shard of the database, the configuration information in the config server is up-to-date, while the cached configuration information in mongos becomes outdated.MongoDB provides the command: flushRouterConfig to force mongos to obtain the latest configuration information from the config server and refresh the mongos cache.

db.adminCommand({"flushRouterConfig":1})}}

Second, shard metadata

Do not directly go to the config server to view the metadata information of the sharded cluster, these data are very important, the safe way is to connect to the config data through mongos or use the sh auxiliary function to view.

Use the sh auxiliary function to view

sh.status()

Connected to mongos to view the collections in the config database

mongos> use config

1，the shards collection stores shard information

db.shards.find()

The data of the shard is stored in the replica set or standalone mongod specified by the host.

{
 "_id" : "shard_name",
 "host" : "replica_set_name"/host:port",
 "tag":[shard_tag1,shard_tag2] 
}

2，the databases collection stores information about all databases in the sharded cluster, regardless of whether the database is sharded

db.databases.find()

If sh.enableSharding(“db_name”) is executed on the database, the value of the partitioned field is true; the primary field specifies the primary shard of the database (primary shard).

{
 "_id" : "test",
 "primary" : "rs0",
 "partitioned" : true
}

3，the collections collection stores information about all sharded collections, excluding unsharded collections (un-sharded collections）

key is: the shard key of the shard

db.collections.find()
{
 "_id" : "test.foo",
 "lastmodEpoch" : ObjectId("57dcd4899bd7f7111ec15f16")
 "lastmod" : ISODate("1970-02-19T17:02:47.296Z"),
 "dropped" : false,
 "key" : {
  "_id" : { 1
 },
 "unique" : true
}

4，the chunks collection stores information about data blocks

ns：the collection of shards, structured as: db_name.collection_name

min and max：the minimum and maximum values of the shard key

shard：the shard where the chunk is located

db.chunks.find()
{
 "_id" : "test.foo"-_id_MinKey",
 "lastmod" : Timestamp(1, 1),
 "lastmodEpoch" : ObjectId("57dcd4899bd7f7111ec15f16")
 "ns" : "test.foo",
 "min" : {
  "_id" : { 1
 },
 "max" : {
  "_id" : { 3087
 },
 "shard" : "rs0"
}

5The changelog collection records operations in the sharded cluster, including chunk splitting and migration operations, as well as the addition or deletion of shards

The 'what' field indicates the type of operation, for example: multi-Split indicates the splitting of chunks

"what" : "addShard"
"what" : "shardCollection.start"
"what" : "shardCollection.end" 
"what" : "multi"-,

6The tags record the shard's tag and the corresponding range of shard keys

{
 "_id" : { "ns" : "records.users", "min" : { "zipcode" : "10001}
 "ns" : "records.users",
 "min" : { "zipcode" : "10001}
 "max" : { "zipcode" : "10281}
 "tag" : "NYC"
}

7The settings collection records the balancer's status and the size of chunks, with the default chunk size being64MB.

{"_id" : "chunksize", "value" :} 64 }
{"_id" : "balancer", "stopped" : false }

8The locks collection records the distributed lock (distributed lock), ensuring that only one mongos instance can perform administrative tasks in the sharded cluster.

When acting as a balancer, mongos acquires a distributed lock and inserts a document into config.locks.

The locks collection stores a distributed lock, ensuring that only one mongos instance can perform administrative tasks on the cluster at once. The mongos acting as balancer acquires a lock by inserting a document similar to the following into the locks collection.

{
 "_id" : "balancer"
 "process" : "example.net:40000:1350402818:16807",
 "state" : 2,
 "ts" : ObjectId("507daeedf40e1879df62e5f3")
 "when" : ISODate("2012-10-16T19:01:01.593Z"),
 "who" : "example.net:40000:1350402818:16807:Balancer:282475249",
 "why" : "doing balance round"
}

Third, delete the shard

When deleting a shard, it must be ensured that the data on the shard is moved to other shards, for sharded collections, use the balancer to migrate data blocks, and for non-sharded collections, the primary shard of the collection must be modified.

1Delete the data of the sharded collection

step1Ensure that the balancer is enabled

sh.setBalancerState(true);

step2migrating all the sharded collections to other shards

use admin
db.adminCommand({"removeShard":"shard_name"})

The removeShard command will migrate data blocks from the current shard to other shards, if there are many data blocks on the shard, the migration process may take a long time.

step3Check the status of data block migration

use admin
db.runCommand( { removeShard: "shard_name" } )

The use of the removeShard command can check the status of data block migration, the "remaining" field indicates the number of remaining data blocks

{
  "msg" : "draining ongoing",
 "state" : "ongoing",
 "remaining" : {
  "chunks" : 42,
  "dbs" : 1
 },
 "ok" : 1
}

step4The data block migration is completed

use admin
db.runCommand( { removeShard: "shard_name" } )
{
 "msg" : "removeshard completed successfully",
 "state" : "completed",
 "shard" : "shard_name",
 "ok" : 1
}

2, delete unsharded databases

step1, view unsharded databases

Unsharded databases include two parts:

　　　　 1, the database has not been sharded, the data has not used sh.enableSharding(\"db_name\"), in the database config, the partitioned field of this database is false

　　　　 2, there are collections in the database that have not been sharded, that is, the current sharding is the primary shard of the collection

use config
db.databases.find({$or:[{"partitioned":false},{"primary":"shard_name"}]})

For databases with partitioned=false, all data is stored in the current shard; for databases with partitioned=true and primary=\"shard_name\", it indicates the existence of un-If a sharded collection is stored in this database, the primary shard of these collections must be changed.

step2, modify the primary shard of the database

db.runCommand( { movePrimary: "db_name", to: "new_shard" })

Fourth, add sharding

Since sharding stores only a part of the dataset, to ensure high availability of data, it is recommended to use a Replica Set as a shard, even if the Replica Set contains only one member. Connect to mongos and use the sh auxiliary function to add sharding.

sh.addShard("replica_set_name"/host:port")

It is not recommended to use standalone mongod as a shard

sh.addShard("host:port")

Fifth, jumbo chunks

In some cases, chunks may continue to grow beyond the chunk size limit, becoming jumbo chunks (jumbo chunk), and the reasons for jumbo chunks are that all documents in the chunk use the same shard key, causing MongoDB to be unable to split the chunk. If this chunk continues to grow, it will lead to uneven distribution of chunks and become a performance bottleneck.

There are restrictions during chunk migration: the size of each chunk cannot exceed2.510,000 documents, or1.3Times the configured value. The default configuration value for chunk size is64MB, chunks exceeding the limit will be marked as jumbo chunks by MongoDB, and MongoDB cannot migrate jumbo chunks to other shards.

MongoDB cannot move a chunk if the number of documents in the chunk exceeds either 250000 documents or 1.3 times the result of dividing the configured chunk size by the average document size.

1, view large chunks

Using sh.status(), you can find large chunks, and there is a jumbo flag behind the large chunks

 { "x" : 2 } -->> { "x" : 3 }) on : shard-a Timestamp(2, 2) jumbo

2, distribute large chunks

Large chunks cannot be split and cannot be automatically distributed through the balancer. They must be manually distributed.

step1, disable balancer

sh.setBalancerState(false)}

step2, increase the configuration value of Chunk Size

Since MongoDB does not allow the movement of large chunks that exceed the size limit, it is necessary to temporarily increase the chunk size configuration value and then evenly distribute the large chunks to the shard cluster.

use config
db.settings.save({"_id":"chunksize","value":"1024"})

step3, move large chunk

sh.moveChunk("db_name.collection_name",{sharded_filed:"value_in_chunk"},"new_shard_name")

step4, enable balancer

sh.setBalancerState(true)

step5, refresh mongos's configuration cache

Force mongos to synchronize configuration information from the config server and refresh the cache.

use admin
db.adminCommand({ flushRouterConfig: 1 })

Six, balancer

The balancer is transformed by mongos, which means that mongos is not only responsible for routing queries to the corresponding shard but also for balancing data blocks. Generally, MongoDB automatically handles data balancing, and the balancer's status can be viewed through config.settings or through the sh auxiliary function

sh.getBalancerState()

Returns true, indicating that the balancer is running, and the system automatically handles data balancing. Using the sh auxiliary function can turn off the balancer

sh.setBalancerState(false)}

The balancer cannot immediately terminate the ongoing block migration operation. When mongos becomes a balancer, it will apply for a balancer lock, check the config.locks collection,

use config
db.locks.find({"_id":"balancer"})
--or 
sh.isBalancerRunning()

if state=2means that the balancer is currently active. If state=0, it means that the balancer has been closed.

The balancing process actually involves migrating data blocks from one shard to another, or first splitting a large chunk into smaller chunks, and then migrating the smaller chunks to other shards. Both the migration and splitting of blocks will increase the system's I/O load. It is best to limit the active time of the balancer to when the system is idle, and you can set the active time window of the balancer to limit the splitting and migration operations of data blocks to a specified time interval.

use config
db.settings.update(
　　{"_id":"balancer"},
　　"$set":{"activeWindow":{"start":"23:00","stop":"04:00"}}),
　　true
)

The split and moved objects of the balancer are chunks, the balancer only guarantees that the number of chunks is balanced on each shard, as for the number of documents contained in each chunk, it is not necessarily balanced. There may be some chunks containing a large number of documents, while some chunks contain a small number of documents, or even no documents at all. Therefore, one should be cautious in choosing the shard index key, that is, the shard key. If a field can meet the vast majority of query needs and also make the distribution of document numbers uniform, then this field is the best choice for the shard key.

Summary

That's all for this article, I hope it can bring some help to your learning or work. If you have any questions, you can leave comments for communication.