MongoDB sharded cluster

4.1. Sharding cluster mechanism and principle

4.1.1. Common MongoDB deployment architecture

Single node: used for development and testing (about 20% of the total)

Replica set: high availability (overall accounted for about 70%)

Sharded cluster: horizontal expansion (overall accounting for about 10%)

4.1.2. Why use sharded cluster

It is recommended to use sharded clusters in the following scenarios:

• MongoDB data capacity is increasing day by day, and access performance is decreasing day by day.

• The system launch is extremely popular and needs to support more concurrent users

• MongoDB already has 10TB of data. A failure occurred and the recovery time was long.

• The system’s access users are global (geographic distribution data is required)

4.1.3. Sharded cluster data distribution method

4.1.3.1. Based on scope

Divide fragments according to the range of data (as shown in the figure below, min~-75, -75 ~ 25, 25 ~175 are all ranges)

Advantages: Using range queries has good performance and can optimize business read operations.

Disadvantages: The data distribution is uneven (prone to hot spots). For example, if the range is divided by the primary key, and the primary key is an auto-incrementing ID, then a large number of write operation data can easily fall into one partition. film, leading to hot issues.

4.1.3.2. Hash-based

Fragment according to the hash value of the data

Advantages: The data is evenly distributed and optimized for write operations. It is suitable for high-concurrency scenarios such as logs and the Internet of Things.

Disadvantages: Range query efficiency is low

4.1.3.3. Custom Zone

Define a Zone based on the node. For example, the international area code: those starting with 1 can read and write servers in the United States, and those starting with 86 can read and write servers in China, which facilitates the deployment of internationalized global projects.

4.1.4. Complete sharded cluster

Routing node (mongos)
Provide a single entrance to the cluster (there can be multiple, in theory only one needs to be used, it is recommended to use at least 2 to achieve high availability)

Function: forward application-side requests, select appropriate data nodes for reading and writing, and merge returns from multiple data nodes

Configuration (Directory) Node

Construction: It is an ordinary replica set architecture, generally 1 master and 2 slaves provide high availability.

Provides cluster metadata storage and mapping of sharded data distribution (which data is placed in which sharded cluster)

The most important thing in configuring the node is the Shared table, which stores the range of data in the shards.

(When mongos starts, it will load the data of the configuration node into its own memory, so that it can quickly and easily compare the data and complete the data distribution and processing)

Data node (mongod)

• Based on replica sets (to avoid single points of failure)

• No data duplication between shards

• Scale out

• Maximum 1024 shards

• All shards work together only

4.1.5. Characteristics of sharded clusters

• The application is completely transparent, no special treatment

• Data automatic balancing

• Dynamic expansion, no need to go offline

• Based on three sharding methods

Summary: Sharded clusters can effectively solve performance bottlenecks and system expansion problems!

Fragmentation consumes more extra money and makes management complicated. Try not to fragment if possible!

4.2. How to make good use of sharded clusters

4.2.1. Reasonable architecture

Shard size

Basic criteria for sharding:

• Data: The amount of data should not exceed 3TB, and try to keep it within 2TB per slice

• Index: Commonly used indexes must be accommodated in memory

After initially determining the shards according to the above standards, you also need to consider the business pressure. As the pressure increases and any of the CPU, RAM, and disk bottlenecks appear, it can be solved by adding more shards.

How to divide

General sharding can be divided according to the following rules:

• Number of shards = total required storage/single server capacity 10TB /2TB =5

• Number of shards=working set size/single server memory capacity 400GB/(256G * 0.6)=3

• Number of shards = Total concurrency/(Single server concurrency0.7) 30000/ (9000 * 0.7) =6

The number of shards is the maximum of the above three: 6

4.2.2. Correct posture

Concepts in sharded clusters

? Shard key: A field in the document
? Document doc: A row of data containing the shard key
? Chunk: Contains n documents
? Shard Shard: Contains n chunks
? Cluster: Contains n shards

Main factors for shard key efficiency

? Value base (Cardinality)
? Value distribution;
? Scattered writing, centralized reading;
? Used in as many business scenarios as possible;
? Avoid monotonically increasing or decreasing shard keys;

A system fragmentation case (email system)

{
_id: ObjectId(),
user: 123,
time: Date(),
subject: "...",
recipients: [],
body: "...",
attachments: []
}

Shard key: { _id: 1}

This primary key sharding method: the cardinality is okay, but the write distribution is uneven, and directional queries cannot satisfy it.

Fragment key: { _id: ”hashed”}

This primary key hash sharding method: the cardinality is okay, but the write distribution is even, but it cannot meet the needs of directed queries.

Shard key: { user_id: 1}

This sharding method of primary key user_id: poor cardinality, even write distribution, and high efficiency for directional queries.

Shard key: { user_id: 1, time:1}

This sharding method of primary key user_id + time is optimal: the cardinality is good, the write distribution is even, and the demand for directed queries is highly efficient.

4.2.3. Sufficient resources

mongos and config usually consume very few resources, so you can choose a low-specification virtual machine;

The resources focus on shard servers:

? Requires enough memory to accommodate hot data indexes;
? When indexes are created correctly the CPU usually does not become a bottleneck unless very heavy computations are involved;
? Try to use SSD as the disk;

Even if sufficient resources are available at the beginning of the project, expansion still needs to be considered when the time is right. It is recommended to monitor the usage of various resources. No matter which one reaches more than 60%, start considering expansion.

• Expansion requires new resources, and applying for new resources takes time.

• The data needs to be balanced after expansion, and balancing takes time. It should be ensured that the speed of new data entering the database is slower than the equilibrium speed

• Balancing requires resources. If resources are about to be or have been exhausted, balancing will be very inefficient.

4.3. Sharded cluster construction and expansion

Learn how to build a 2-shard sharded cluster

Environmental requirements: 3 Linux virtual machines (4 cores 8G recommended)

Overall actual combat process

• Configure domain name resolution

• Prepare shard directory

• Create the first shard replica set and initialize it

• Create config replica set and initialize it

• Initialize the sharded cluster and add the first shard

• Create sharded table

• Add a second shard (simulated expansion)

4.3.1. Preparation

1. Configure domain name resolution

Execute the following three commands on the three virtual machines respectively. Be careful to replace the actual IP address.

echo "192.168.1.25 demo1 member1.msb.com member2.msb.com" >> /etc/hosts
echo "192.168.1.26 demo2 member3.msb.com member4.msb.com" >> /etc/hosts
echo "192.168.1.27 demo3 member5.msb.com member6.msb.com" >> /etc/hosts

2. Prepare the shard directory

Create a data directory on each server. We use /data. Please change it to other directories according to your needs.

Execute the following commands on member1/member3/member5:

mkdir -p /data/shard1/
mkdir -p /data/config/

Execute the following commands on member2/member4/member6:

mkdir -p /data/shard2/
mkdir -p /data/mongos/

4.3.2. Build shards

1. Build shard1

Execute the following commands on member1/member3/member5.

mongod --bind_ip 0.0.0.0 --replSet shard1 --dbpath /data/shard1 --logpath /data/shard1/mongod.log --port 27010 --fork --shardsvr --wiredTigerCacheSizeGB 1

Note the following parameters:

• shardsvr: Indicates that this is not an ordinary replica set, but part of a shard set;

• wiredTigerCacheSizeGB: This parameter indicates the cache size that MongoDB can use. The default value is (RAM - 1GB) / 2.

• • It is not recommended to configure more than the default value, as there is a risk of OOM;

• • Because our current test will run multiple instances on one server, a smaller value is configured;

• bind_ip: It is strongly recommended not to bind external IP addresses in a production environment. All IP addresses are bound here for the convenience of demonstration. For a similar reason, authentication --auth should be turned on in the production environment, but it is not used here for demonstration convenience;

Use these three examples to build a shard1 replica set:

• Connect to any instance, for example we connect to member1.msb.com:

mongo --host member1.msb.com:27010

• Initialize the shard1 replica set. We initialize the replica set with the following configuration:

rs.initiate({
    _id: "shard1",
    "members" : [
        {
            "_id": 0,
            "host" : "member1.msb.com:27010"
        },
        {
            "_id": 1,
            "host" : "member3.msb.com:27010"
        },
        {
            "_id": 2,
            "host" : "member5.msb.com:27010"
        }
    ]
});

2. Build config

Similar to shard1, we can build a config server. Execute the following commands on member1/member3/member5:

• Run the config instance:

mongod --bind_ip 0.0.0.0 --replSet config --dbpath /data/config --logpath /data/config/mongod.log --port 27019 --fork --configsvr --wiredTigerCacheSizeGB 1

• Connect to member1:

mongo --host member1.msb.com:27019

• Initialize config replica set:

rs.initiate({
    _id: "config",
    "members" : [
        {
            "_id": 0,
            "host" : "member1.msb.com:27019"
        },
        {
            "_id": 1,
            "host" : "member3.msb.com:27019"
        },
        {
            "_id": 2,
            "host" : "member5.msb.com:27019"
        }
    ]
});

3. Build mongos

The construction of mongos is relatively simple. We build 3 mongos on member2/member4/member6. Note the following parameters:

• configdb: Indicates the cluster address used by config;

Start building:

• Run the mongos process:

mongos --bind_ip 0.0.0.0 --logpath /data/mongos/mongos.log --port 27017 --configdb config/member1.msb.com:27019,member3.msb.com:27019,member5.msb. com:27019 --fork

• Connect to any mongos, here we use member1:

mongo --host member1.msb.com:27017

• Add shard1 to the cluster:

sh.addShard("shard1/member1.msb.com:27010,member3.msb.com:27010,member5.msb.com:27010");

4. Test shard set

In the above example, we built a shard set with only 1 shard. Before continuing let’s test this shard set.

• Connect to shard set:

mongo --host member1.msb.com:27017

sh.status();

• Create a sharded table:

sh.enableSharding("foo");
sh.shardCollection("foo.bar", {_id: 'hashed'});
sh.status();

• Write any amount of data:

use foo
for (var i = 0; i < 10000; i + + ) {
    db.bar.insert({i: i});
}

5. Add new shards to the shard set

Next, we build shard2 and add it to the shard set to observe the effect.

Use a method similar to shard1 to build shard2. Execute the following commands on member2/member4/member6:

mongod --bind_ip 0.0.0.0 --replSet shard2 --dbpath /data/shard2 --logpath /data/shard2/mongod.log --port 27011 --fork --shardsvr --wiredTigerCacheSizeGB 1

Use these three examples to build a shard2 replica set:

• Connect to any instance, for example we connect to member2.msb.com:

mongo --host member2.msb.com:27011

• Initialize the shard2 replica set. We initialize the replica set with the following configuration:

rs.initiate({
    _id: "shard2",
    "members" : [
        {
            "_id": 0,
            "host" : "member2.msb.com:27011"
        },
        {
            "_id": 1,
            "host" : "member4.msb.com:27011"
        },
        {
            "_id": 2,
            "host" : "member6.msb.com:27011"
        }
    ]
});

• Connect to any mongos. member1 is used here:

mongo --host member1.msb.com:27017

• Add shard2 to the cluster:

sh.addShard("shard2/member2.msb.com:27011,member4.msb.com:27011,member6.msb.com:27011");

• Observe sh.status():

sh.status();

It can be found that the two chunks originally on shard1 have been balanced to shard2. This is MongoDB’s automatic balancing mechanism.