MongoDB Sharding: Concepts, Examples & Tutorials

This comprehensive article explores sharding in MongoDB. We break the article into two sections:

  1. Concepts, components, pros & cons
  2. A step-by-step tutorial on setting up sharding

(This article is part of our MongoDB Guide. Use the right-hand menu to navigate.)

What is sharding?

Sharding is the process of distributing data across multiple hosts. In MongoDB, sharding is achieved by splitting large data sets into small data sets across multiple MongoDB instances.

How sharding works

When dealing with high throughput applications or very large databases, the underlying hardware becomes the main limitation. High query rates can stress the CPU, RAM, and I/O capacity of disk drives resulting in a poor end-user experience.

To mitigate this problem, there are two types of scaling methods.

Vertical scaling

Vertical scaling is the traditional way of increasing the hardware capabilities of a single server. The process involves upgrading the CPU, RAM, and storage capacity. However, upgrading a single server is often challenged by technological limitations and cost constraints.

Horizontal scaling

This method divides the dataset into multiple servers and distributes the database load among each server instance. Distributing the load reduces the strain on the required hardware resources and provides redundancy in case of a failure.

However, horizontal scaling increases the complexity of underlying architecture. MongoDB supports horizontal scaling through sharding—one of its major benefits, as we’ll see below.

MongoDB sharding basics

MongoDB sharding works by creating a cluster of MongoDB instances consisting of at least three servers. That means sharded clusters consist of three main components:

Shard

A shard is a single MongoDB instance that holds a subset of the sharded data. Shards can be deployed as replica sets to increase availability and provide redundancy. The combination of multiple shards creates a complete data set. For example, a 2 TB data set can be broken down into four shards, each containing 500 GB of data from the original data set.

Mongos

Mongos act as the query router providing a stable interface between the application and the sharded cluster. This MongoDB instance is responsible for routing the client requests to the correct shard.

Config Servers

Configuration servers store the metadata and the configuration settings for the whole cluster.

Components illustrated

The following diagram from the official MongoDB docs explains the relationship between each component:

  1. The application communicates with the routers (mongos) about the query to be executed.
  2. The mongos instance consults the config servers to check which shard contains the required data set to send the query to that shard.
  3. Finally, the result of the query will be returned to the application.

It’s important to remember that the config servers also work as replica sets.

Shard Keys

When sharding a MongoDB collection, a shard key gets created as one of the initial steps. The “shard key” is used to distribute the MongoDB collection’s documents across all the shards. The key consists of a single field or multiple fields in every document. The sharded key is immutable and cannot be changed after sharding. A sharded collection only contains a single shard key.

When sharding a populated collection, the collection must have an index that starts with the shard key. For empty collections that don’t have an appropriate index, MongoDB will create an index for the specified shard key.

The shard key can directly have an impact on the performance of the cluster. Hence can lead to bottlenecks in applications associated with the cluster. To mitigate this, before sharding the collection, the shard key must be created based on:

Chunks

Chunks are subsets of shared data. MongoDB separates sharded data into chunks that are distributed across the shards in the shared cluster. Each chunk has an inclusive lower and exclusive upper range based on the shard key. A balancer specific for each cluster handles the chunk distribution.

The balancer runs as a background job and distributes the chunks as needed to achieve an even balance of chunks across all shards. This process is called even chuck distribution.

Sharding benefits & limitations

Now that we’ve got the concept down, let’s look at benefits and limitations of sharding in MongoDB:

Benefits

Limitations

How to set up a MongoDB sharding

Now that we understand the concepts of sharding, let’s get started with some tutorials.

The following steps demonstrate how to set up a sharded cluster can on three servers. Ubuntu 20.04 LTS and MongoDB 4.4.1 are installed on all three servers. We can define servers as shown below:

Now, we’ll walk you through several actions, including how to:

Configuring the config server

Create the directory structure for the database and log file.

mkdir -pv  mongodb/data/configdb/
mkdir -pv  mongodb/data/logs
touch mongodb/data/logs/configsvr.log
ls -alRv mongodb/

Configuring the config serverCreate the config file. When you execute the above commands, a config file will be created along with those directories. Next, open the config file and assign the port and the IP address of the server. Also, don’t forget to set the cluster role as configsvr.

sudo nano /etc/mongodConfig.conf

File - mongodConfig.conf

storage:
  dbPath: /home/barry/mongodb/data/configdb
  journal:
  enabled: true

systemLog:
  destination: file
  logAppend: true
  path: /home/barry/mongodb/data/logs/configsvr.log

net:
  port: 27019
  bindIp: 10.10.10.56

sharding:
  clusterRole: configsvr

replication:
  replSetName: ConfigReplSet

Start the config server. Now, you can start the config server with the following command.

mongod --config /etc/mongodConfig.conf&

Configuring the config server 2Check the logs to verify if the server is running.

tail -100 mongodb/data/logs/configsvr.log

Configuring the config server 3Connect to the config server.

mongo 10.10.10.56:27019

Configuring the config server 4Initiate the Config Server. You can use the initiate() function to initiate the config server with the default configuration. Then check the status of the server with the status() function.

rs.initiate()
rs.status()

Configuring the config server 5

Configuring the query router

Create the directory structure and an empty log file.

mkdir -pv  mongodb/data/logs
touch mongodb/data/logs/mongorouter.log
ls -alRv mongodb/

Configuring the query routerCreate the config file. To create the query router, we provide the log location, IP, and port of the server instance. Further, we need to define the config server to which this query router belongs.

sudo nano /etc/mongoRouter.conf

File - mongoRouter.conf

systemLog:
  destination: file
  logAppend: true
  path: /home/barry/mongodb/data/logs/queryrouter.log

net:
  port: 27017
  bindIp: 10.10.10.57

sharding:
  configDB: ConfigReplSet/10.10.10.56:27019

Start the query router with the mongos command.

mongos --config /etc/mongoRouter.conf&

Check if the query router instance is reachable.

mongo 10.10.10.57:27017

Configuring the query router 2

Configuring a shard

Create the directory structure and the log file.

mkdir -pv  mongodb/data/sharddb/
mkdir -pv  mongodb/data/logs
touch mongodb/data/logs/shard.log
ls -alRv mongodb/

Configuring a shardCreate the config file. The shard’s config file contains the paths for the database storage, logs, and sharding cluster role, which is set to shardsvr.  It also includes the network settings to the server instance. Finally, we have set replSetName allowing the data to be replicated.

sudo nano /etc/mongodShard.conf

File - mongoShard.conf

storage:
  dbPath: /home/barry/mongodb/data/sharddb
  journal:
    enabled: true

systemLog:
  destination: file
  logAppend: true
  path: /home/barry/mongodb/data/logs/shard.log

net:
  port: 27018
  bindIp: 10.10.10.58

sharding:
  clusterRole: shardsvr

replication:
  replSetName: ShardReplSet

Start the shard server.

mongod --config /etc/mongodShard.conf&

Configuring a shard 2Check the logs to verify if the server is running.

tail -100 mongodb/data/logs/shard.log

Configuring a shard 3Login to the shard server.

mongo 10.10.10.58:27018

Configuring a shard 4Initiate the shard server. The replication with the default configuration is enabled using the initiate() function. Then check the status of the initialization with the status() function.

rs.initiate()
rs.status()

Configuring a shard 5

Adding the shard to the Cluster

Connect to the Query Router.

mongo 10.10.10.57:27017

Add the Shard to the clustet. Using the addShard() command, we provide the replSetName with the IP address and the port of the shard instance.

sh.addShard( "ShardReplSet/10.10.10.58:27018")

Adding the shard to the Cluster

Enabling the sharding

Create the database. A database named “persons” will be created using the query router. This database will be used in the sharding operation. Here the necessary steps to enable sharding:

use persons
sh.enableSharding("persons")

Adding the shard to the Cluster 2Check sharding status. Using the status() command, we check if the sharding is enabled to the database. As shown below, sharding is enabled on the “persons” database.

sh.status()

Configuring the config server 5 Adding the shard to the Cluster 3

Creating the sharding dataset

In this section, we will create a MongoDB collection and an index called person_id to be used as the shard key. The data in the collection will be sharded using the collection and the shard key.

Create the collection.

db.createCollection("personscollection")

Creating the sharding datasetCreate the index and add a record. Create the index with personid as the field in descending order.

db.personscollection.createIndex({personid: -1})

Add a personid with value 10001.

db.personscollection.insertOne({personid: 10001})

Creating the sharding dataset 2

Enabling sharding for the “personscollection”

The first step is to make sure the index personid is hashed. If not, this will result in an error, and sharding will fail.

To achieve this, we use ensureIndex() function within MongoDB.

db.personscollection.ensureIndex({personid : "hashed"})

If the index is used correctly, we can enable Sharding with “personid” as the shard key.

sh.shardCollection("persons.personscollection", {personid : "hashed"})

If the operation is successful, the following output will be displayed.

Creating the sharding dataset 4Verify if sharding is working as intended. Use the getShardDistribution() command to verify the status of the sharding operation.

db.personscollection.getShardDistribution()

Creating the sharding dataset 4The above output describes that the personscollection is sharded in the ShardRepSet on the Shard server (10.10.10.58). It consists of a single document in a single chunk. The available document is the single record we entered into the collection.

Summing up MongoDB sharding

MongoDB sharding is a method to manage large data sets efficiently by distributing the workload across many servers without having any adverse effects on the overall performance of the database. Also, sharding provides the ability to efficiently scale the cluster for future requirements without a complex restructuring of the underlying hardware infrastructure.

Related reading