← Back to list

MongoDB Sharded Cluster: A Step by Step Implementation from Scratch

MongoDB is a document-oriented NoSQL database designed for flexibility, scalability, and high performance. Instead of storing data in…

Mai Hoang Viet · 2025-12-01 15:54 · 0 claps · 7.7 min read
#mongodb #devops #mongodb-sharding #high-availability #scalability
Open on Medium ↗
Wiki topics: ☁️ · DevOps & Cloud

MongoDB Sharded Cluster: A Step by Step Implementation from Scratch

MongoDB is a document-oriented NoSQL database designed for flexibility, scalability, and high performance. Instead of storing data in traditional rows and tables, MongoDB uses JSON-like documents, making it well-suited for modern applications that evolve quickly and handle diverse data structures.

As datasets grow, a single server can’t always keep up with storage, read/write throughput, or fault-tolerance requirements. This is where a MongoDB Sharded Cluster comes in. Sharding allows MongoDB to distribute data across multiple machines, enabling horizontal scaling, balanced workloads, and predictable performance even at very large data volumes.

A sharded cluster typically consists of three main components: config servers for metadata, shards for actual data storage (each implemented as a replica set), and mongos routers that route client queries to the appropriate shard. When combined, this architecture allows the database to grow linearly and operate reliably even under heavy load.

Common use cases for MongoDB sharding include high-volume OLTP systems, real-time analytics platforms, event logging pipelines, time-series data, IoT workloads, and any application where a single server cannot handle the dataset size or throughput demands.

Implementation Architecture

Install MongoDB

Prerequisite

4 RHEL 8 x64 servers.

# Add host
sudo tee -a /etc/hosts <<EOF
192.168.48.1 MDB-SERVER-01
192.168.48.2 MDB-SERVER-02
192.168.48.3 MDB-SERVER-03
192.168.48.4 MDB-MONGOS-SERVER
EOF

# Disable SELinux
setenforce 0
sed -i 's/^SELINUX=enforcing/SELINUX=disabled/' /etc/selinux/config

# Disable firewalld
systemctl stop firewalld
systemctl disable firewalld

# Reboot server
reboot

Add MongoDB 8.0 repo

cat <<EOF | tee /etc/yum.repos.d/mongodb-org-8.0.repo
[mongodb-org-8.0]
name=MongoDB Repository
baseurl=https://repo.mongodb.org/yum/redhat/8/mongodb-org/8.0/x86_64/
gpgcheck=1
enabled=1
gpgkey=https://pgp.mongodb.com/server-8.0.asc
EOF

Installation

yum install -y mongodb-org

Disable the default service

Since we’ll be running multiple custom instances (configsvr, shardsvr, mongos), the default MongoDB service must be disabled.

systemctl stop mongodb.service
systemctl disable mongodb.service

Create data and log directories for each role:

sudo mkdir -p /data/mongodb/data/configsvr
sudo mkdir -p /data/mongodb/data/shardsvr-01
sudo mkdir -p /data/mongodb/data/shardsvr-02
sudo mkdir -p /data/mongodb/data/shardsvr-03

sudo mkdir -p /data/mongodb/log

Configure MongoDB

Config Server

Config Server Configuration

Create the configuration file:

cat <<'EOF' > /etc/configsvr.conf
sharding:
   clusterRole: configsvr

replication:
   replSetName: configsvr

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

storage:
   dbPath: /data/mongodb/data/configsvr

net:
   bindIp: 0.0.0.0
   port: 27020

setParameter:
   enableLocalhostAuthBypass: false
EOF

Create the systemd service:

cat <<'EOF' > /etc/systemd/system/configsvr.service
[Unit]
Description=MongoDB Config Server
After=network.target

[Service]
User=root
ExecStart=/usr/bin/mongod --config /etc/configsvr.conf
PIDFile=/var/run/mongodb/mongod.pid
Restart=on-failure
LimitNOFILE=64000
Type=simple

[Install]
WantedBy=multi-user.target
EOF

Start the configsvr service:

systemctl daemon-reload

systemctl enable configsvr.service
systemctl start configsvr.service
systemctl status configsvr.service

Initialize the Config Server Replica Set

On MDB-SERVER-01, connect to configsvr:

mongosh --port 27020

Then, configure the replica set:

rs.initiate({
  _id: "configsvr",
  configsvr: true,
  members: [
    { _id: 0, host: "MDB-SERVER-01:27020" },
    { _id: 1, host: "MDB-SERVER-02:27020" },
    { _id: 2, host: "MDB-SERVER-03:27020" },
  ],
});

The result:

configsvr [direct: primary] test> rs.status().members.map(m => ({
...   name: m.name,
...   stateStr: m.stateStr,
...   health: m.health
... }))
...
[
  { name: 'MDB-SERVER-01:27020', stateStr: 'PRIMARY', health: 1 },
  { name: 'MDB-SERVER-02:27020', stateStr: 'SECONDARY', health: 1 },
  { name: 'MDB-SERVER-03:27020', stateStr: 'SECONDARY', health: 1 }
]
configsvr [direct: primary] test>

Shard

Shard Configuration

The following configuration and systemd service templates apply to all shard replica set members. Only the shard name and port are unique per instance. All other values, including replica set name, data directory, and log path, are derived directly from the shard name.

Apply the templates below using the following shard names and ports:

  • shardsvr-01 → port 27021
  • shardsvr-02 → port 27022
  • shardsvr-03 → port 27023

Shard configuration template:

cat <<'EOF' > /etc/<shard-name>.conf
storage:
  dbPath: /data/mongodb/data/<shard-name>

systemLog:
  destination: file
  logAppend: true
  path: /data/mongodb/log/<shard-name>.log

net:
  port: <port>
  bindIp: 0.0.0.0

replication:
  oplogSizeMB: 50
  replSetName: <shard-name>

sharding:
  clusterRole: shardsvr
EOF

systemd service template (Shard):

cat <<'EOF' > /etc/systemd/system/<shard-name>.service
[Unit]
Description=MongoDB Database Server
Documentation=https://docs.mongodb.org/manual
After=network.target

[Service]
User=root
Group=root
ExecStart=/usr/bin/mongod --config /etc/<shard-name>.conf
PIDFile=/var/run/mongodb/<shard-name>.pid
# file size
LimitFSIZE=infinity
# cpu time
LimitCPU=infinity
# virtual memory size
LimitAS=infinity
# open files
LimitNOFILE=64000
# processes/threads
LimitNPROC=64000
# locked memory
LimitMEMLOCK=infinity
# total threads (user+kernel)
TasksMax=infinity
TasksAccounting=false

# Recommended limits for for mongod as specified in
# http://docs.mongodb.org/manual/reference/ulimit/#recommended-settings

[Install]
WantedBy=multi-user.target
EOF

Enable and start shard services:

systemctl daemon-reload

systemctl enable shardsvr-01.service
systemctl enable shardsvr-02.service
systemctl enable shardsvr-03.service

systemctl start shardsvr-01.service
systemctl start shardsvr-02.service
systemctl start shardsvr-03.service

Initialize Shard Replica Sets

Each shard must be initialized as an independent replica set. The procedure is identical for all shards and only differs in the replica set name and member ports:

  • shardsvr-01 → port 27021
  • shardsvr-02 → port 27022
  • shardsvr-03 → port 27023

Example: Initialize shardsvr-01 :

  • On MDB-SERVER-01, connect to shardsvr-01:
mongosh --port 27021
  • Then, configure the replica set:
rs.initiate({
  _id: "shardsvr-01",
  members: [
    { _id: 0, host: "MDB-SERVER-01:27021" },
    { _id: 1, host: "MDB-SERVER-02:27021" },
    { _id: 2, host: "MDB-SERVER-03:27021" }
  ]
})
  • The result:
shardsvr-01 [direct: primary] test> rs.status().members.map(m => ({
...   name: m.name,
...   stateStr: m.stateStr,
...   health: m.health
... }))
...
[
  { name: 'MDB-SERVER-01:27021', stateStr: 'PRIMARY', health: 1 },
  { name: 'MDB-SERVER-02:27021', stateStr: 'SECONDARY', health: 1 },
  { name: 'MDB-SERVER-03:27021', stateStr: 'SECONDARY', health: 1 }
]

Mongos

Mongos Configuration

Create the configuration file:

cat <<'EOF' > /etc/mongos.conf
systemLog:
  destination: file
  logAppend: true
  path: /var/log/mongodb/mongos.log

net:
  port: 27017
  bindIp: 0.0.0.0

sharding:
  configDB: configsvr/MDB-SERVER-01:27020,MDB-SERVER-02:27020,MDB-SERVER-03:27020
EOF

Create the systemd service:

cat <<'EOF' > /etc/systemd/system/mongos.service
[Unit]
Description=MongoDB Database Server
Documentation=https://docs.mongodb.org/manual
After=network.target

[Service]
User=root
Group=root
ExecStart=/usr/bin/mongos --config /etc/mongos.conf
PIDFile=/var/run/mongodb/mongos.pid
# file size
LimitFSIZE=infinity
# cpu time
LimitCPU=infinity
# virtual memory size
LimitAS=infinity
# open files
LimitNOFILE=64000
# processes/threads
LimitNPROC=64000
# locked memory
LimitMEMLOCK=infinity
# total threads (user+kernel)
TasksMax=infinity
TasksAccounting=false

# Recommended limits for for mongod as specified in
# http://docs.mongodb.org/manual/reference/ulimit/#recommended-settings

[Install]
WantedBy=multi-user.target
EOF

Start mongos service:

systemctl daemon-reload

systemctl enable mongos.service
systemctl start mongos.service
systemctl status mongos.service

Add Shards to the Cluster

On the MDB-MONGOS-SERVER, connect to mongos:

mongosh --port 27017

Add shards to the database:

sh.addShard("shardsvr-01/MDB-SERVER-01:27021")
sh.addShard("shardsvr-02/MDB-SERVER-02:27022")
sh.addShard("shardsvr-03/MDB-SERVER-03:27023")

sh.status()

The result:

[direct: mongos] test> use config
switched to db config
[direct: mongos] config> db.shards.find().pretty()
[
  {
    _id: 'shardsvr-01',
    host: 'shardsvr-01/MDB-SERVER-01:27021,MDB-SERVER-02:27021,MDB-SERVER-03:27021',
    state: 1,
    topologyTime: Timestamp({ t: 1756280717, i: 10 }),
    replSetConfigVersion: Long('2')
  },
  {
    _id: 'shardsvr-02',
    host: 'shardsvr-02/MDB-SERVER-01:27022,MDB-SERVER-02:27022,MDB-SERVER-03:27022',
    state: 1,
    topologyTime: Timestamp({ t: 1756280980, i: 10 }),
    replSetConfigVersion: Long('2')
  },
  {
    _id: 'shardsvr-03',
    host: 'shardsvr-03/MDB-SERVER-01:27023,MDB-SERVER-02:27023,MDB-SERVER-03:27023',
    state: 1,
    topologyTime: Timestamp({ t: 1756280986, i: 9 }),
    replSetConfigVersion: Long('2')
  }
]

Configure Member Priority for Each Shard

In a sharded cluster, each replica set elects its own primary. By configuring priorities, we ensure the intended node becomes the primary for each shard, improve failover behavior, and avoid the situation where all shard primaries end up on the same node. This helps distribute write load evenly and increases overall cluster stability.

Example: Configure Priority for shardsvr-01 :

  • On MDB-SERVER-01, connect to shardsvr-01:
mongosh --port 27021
  • Get the member of shardsvr-01:
shardsvr-01 [direct: primary] test> rs.conf().members.map(m => ({ _id: m._id, host: m.host }))
[
  { _id: 0, host: 'MDB-SERVER-01:27021' },
  { _id: 1, host: 'MDB-SERVER-02:27021' },
  { _id: 2, host: 'MDB-SERVER-03:27021' }
]
  • Then, configure shardsvr-01 on MDB-SERVER-01 as the primary:
cfg = rs.conf()

cfg.members[0].priority = 2   // primary priority
cfg.members[1].priority = 1
cfg.members[2].priority = 1

rs.reconfig(cfg)
rs.stepDown()
  • The result:
shardsvr-01 [direct: primary] test> rs.conf().members.forEach(m => print(m.host + " → priority: " + m.priority))
...
MDB-SERVER-01:27021 → priority: 2
MDB-SERVER-02:27021 → priority: 1
MDB-SERVER-03:27021 → priority: 1

Repeat the same procedure for shardsvr-02 and shardsvr-03, adjusting member priorities according to the architecture described above.

Create Database and Configure Shard for Collections

In MongoDB, databases and collections are created implicitly when you insert data.

On MDB-MONGOS-SERVER, connect to mongos:

mongosh --port 27017

Create the sales database with the orders collection:

use sales
db.createCollection("orders")

To enable a database to contain sharded collections, you must enable sharding for that database:

sh.enableSharding("sales")

If you do not run this command, you will not be able to shard any collections in the database.

Next, select the shard key and shard the collection.

Shard by the orderID field in the orders collection:

sh.shardCollection("sales.orders", { orderId: "hashed" });

An index must exist on the shard key field; if it does not exist, MongoDB will create it automatically.

The shard key cannot be deleted or changed later.

Insert sample data for testing:

let bulk = [];
const totalDocs = 1_000_000;     // total number of documents to insert
const batchSize = 20_000;        // larger batch size reduces round-trips

for (let i = 0; i < totalDocs; i++) {
  bulk.push({
    orderId: i,
    orderDate: new Date(2000 + (i % 20), 0, 1),
    amount: Math.floor(Math.random() * 1000)
  });

  if (bulk.length === batchSize) {
    db.orders.insertMany(bulk, { ordered: false });  // unordered insert
    bulk = [];
  }
}

// Insert the remaining documents
if (bulk.length > 0) {
  db.orders.insertMany(bulk, { ordered: false });
}

print(`✅ Successfully inserted ${totalDocs} documents into sales.orders`);

Check if the collection has been sharded:

[direct: mongos] sales> use config
switched to db config
[direct: mongos] config> db.collections.find({ _id: "sales.orders" }).pretty()
[
  {
    _id: 'sales.orders',
    lastmodEpoch: ObjectId('68ba92058f9ccdd158c62532'),
    lastmod: ISODate('2025-09-05T07:32:21.854Z'),
    timestamp: Timestamp({ t: 1757057613, i: 56 }),
    uuid: UUID('42d42171-9f63-4642-997c-5f83097fc611'),
    key: { orderId: 'hashed' },
    unique: false,
    noBalance: false
  }
]
[direct: mongos] config>

Check the shard status of the database:

[direct: mongos] test> use sales
switched to db sales
[direct: mongos] sales> db.orders.getShardDistribution()
Shard shardsvr-02 at shardsvr-02/MDB-SERVER-01:27022,MDB-SERVER-02:27022,MDB-SERVER-03:27022
{
  data: '21.01MiB',
  docs: 333819,
  chunks: 1,
  'estimated data per chunk': '21.01MiB',
  'estimated docs per chunk': 333819
}
---
Shard shardsvr-03 at shardsvr-03/MDB-SERVER-01:27023,MDB-SERVER-02:27023,MDB-SERVER-03:27023
{
  data: '21.01MiB',
  docs: 333951,
  chunks: 1,
  'estimated data per chunk': '21.01MiB',
  'estimated docs per chunk': 333951
}
---
Shard shardsvr-01 at shardsvr-01/MDB-SERVER-01:27021,MDB-SERVER-02:27021,MDB-SERVER-03:27021
{
  data: '20.91MiB',
  docs: 332230,
  chunks: 1,
  'estimated data per chunk': '20.91MiB',
  'estimated docs per chunk': 332230
}
---
Totals
{
  data: '62.94MiB',
  docs: 1000000,
  chunks: 3,
  'Shard shardsvr-02': [
    '33.38 % data',
    '33.38 % docs in cluster',
    '66B avg obj size on shard'
  ],
  'Shard shardsvr-03': [
    '33.39 % data',
    '33.39 % docs in cluster',
    '66B avg obj size on shard'
  ],
  'Shard shardsvr-01': [
    '33.22 % data',
    '33.22 % docs in cluster',
    '66B avg obj size on shard'
  ]
}
[direct: mongos] sales>

Conclusion

In this guide, we built a complete MongoDB sharded cluster from scratch, consisting of three shard replica sets and a dedicated config server replica set. This setup establishes the core architecture required for horizontal scaling, predictable failover, and balanced data distribution.

At this stage, the cluster is fully functional but intentionally left open. Clients can connect without credentials, and internal communication between mongod and mongos instances relies on implicit trust. This makes the deployment useful for validating sharding behavior and cluster topology, but unsuitable for real environments.

The next step is to harden the cluster by securing internal communication and enforcing authenticated access. In the following article, we will upgrade this sharded cluster to use Keyfile Authentication, transforming it from an open setup into a securely authenticated MongoDB deployment.

Next article: Hardening an Existing MongoDB Sharded Cluster with Keyfile Authentication


메타데이터
post_id
cf0d5ad7f206
slug
building-mongodb-sharded-clusters-the-right-way-cf0d5ad7f206
url
https://medium.com/@maihoangviet/building-mongodb-sharded-clusters-the-right-way-cf0d5ad7f206
canonical_url
https://medium.com/@maihoangviet/building-mongodb-sharded-clusters-the-right-way-cf0d5ad7f206
author_url
https://medium.com/@maihoangviet
status
ok
fetched_at
2026-06-09 15:37:30