MongoDB for DBA's 6/7. Scalability

Chapter 6. SCALABILITY: Sharding setup, sharding monitoring, shard key selection, inserting large amounts of data

Sharding & Data Distribution

Shards : they are partitions where each one has different data (different documents). One document will live on one and only one shard and it is possible for documents to move for rebalancing purposes but in a given point in time the document let someone shard. It is used to scale-out the system data. We can also have a replication having the same documents in different members of a replica set for high availability and its safety and disaster recovery. 


In mongodb, the data distribution is based on a sharded key, and in a given collection the documents that have the same sharded key will be on the same shard member and documents who sharded key are closed to other sharded key because each shard member will have a range of documents (range based partitioning) corresponding with the sharded key. A given range if given key range will live on a particural shard.

Example: sharding key = customer name

1. Shard 0 :  range [A - Irene]
2. Shard 1  :  range [Irene - Maria]
3. Shard 3  :  range [Maria - Robert]
4. Shard 4  :  range [Robert - Zzzzzz]

We can have sharded key for differents collections.

Why range based ?

• The involve rangers will have an efficiency rate.
• The query will be done on a range efficiently 
• Sorting can be quite useful

A field that has the boolean of value of either true or false would make for a good shard key:

TrueFalse

Chunks & Operations

Chunk in mongodb is refered to all the documents in the key range.

Operations with chunks that can have in the background:

1. "Split": split the key range in two different key ranges because the limit of data has exceeded. That will make sure that there are no huge chunks and the data will be moved. This operations is unexpensive.
2. "Migrate": it will do when "the balancer" sees there is a lack of balance between the number of shards on the different shards and decides to do migrations. This operations is more expensive because the data transfer between members. While the data is transfered, the system is still live and reads and writes of documents that were in the key range during the migration will execute.

It is possible to configure the size of each chunk which affects when they will be split. A large chunk size means chunks will be split less often as each chunk needs to grow larger before it is split, and a small chunk size means they will be split more often, for instance.

Which of the following statements are true regarding a very small chunk size configuration as compared to a large chunk size configuration? Check all that apply.

Fewer migrations would have to take placeMore migrations would have to take placeEach shard would be very balanced in terms of how much data they containEach server would be very unbalanced in terms of how much data they containThere is no difference in terms of migration and shard server balance

Sharding Processes

Each shard server will run one or more mongod processes storing data (with replica sets for instance). We will have multiple mongod's procecesses in multimple mongo servers witin a given shard. 

In a addition, we have config servers (small mongod's) which store the metadata of the clusters (chunks of the system). 

There are one more component: the mongos process --> the data relying component. We can have any number of these and our clients will connect them to perform operations on the clusters. The mongos gives a connection from a client, a view of the whole cluster as a single logical entity and it does not need to worry about if a replica set o shard systems exists.

The mongos will talk about to everything as needed and to the config servers to get metadata and when a query comes in from a client, the mongos just started up, it will get a relevant metadata it needs to decide where to send the query from the config servers, and then, it will do the operation with the mongod's. If it comunicates with more than of those, it will merge data coming back on as-needed basis. 

Properties of these different components: 

1. The mongos processes: have no persistent state. There has no data files. There are really kind of like a load balancer and they will get information they need on how to get the data and load it and route it from the config servers and they will cache that in RAM.
2. The mongod's processes: are the data storers (database)
3. The config servers: are the metadata store (contain which shard has the right data). We can run one config server, if we are running in a development on a laptop. There are three config servers in a production mongodb cluster in order to get consensus if one of them get failure. They have identical data with a copy of the metadata that they keep in sync. As long as one config server is up, the cluster is alive. If all three were down, the cluster would be down. If all three are not up, metadata changing operations like splits and migrates cannot happen

db.foo.find( 
  { name : { $gt : "Joe", $lt : "John" } } ) 

is efficient if the collection is sharded on { name : 1 }

TrueFalse

Cluster Topology

A Cluster setup minimun will be:

1. how many shards initially --> 4 shards
2. replication factor --> R = 3

it will have 4x3=12 mongod processes in shard servers running from 4 mongos processes

Running on localhost

Example with 4 shards processes and 3 replica sets in one machine

mkdir a0
mkdir a1
mkdir a2
mkdir b0
mkdir b1
mkdir b2
mkdir c0
mkdir c1
mkdir c2
mkdir d0
mkdir d1
mkdir d2

mkdir cfg0
mkdir cfg1
mkdir cfg2

Run the config servers:

mongod --configsvr --dbpath cfg0 --port 26050 --fork --logpath log.cfg0 --logappend
mongod --configsvr --dbpath cfg1 --port 26051 --fork --logpath log.cfg1 --logappend
mongod --configsvr --dbpath cfg2 --port 26052 --fork --logpath log.cfg2 --logappend

Run the shard servers:

mongod --shardsvr --replSet a --dbpath a0 --logpath log.a0 --port 27000 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet a --dbpath a1 --logpath log.a1 --port 27001 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet a --dbpath a2 --logpath log.a2 --port 27002 --fork --logappend --smallfiles --oplogSize 50

mongod --shardsvr --replSet b --dbpath b0 --logpath log.b0 --port 27100 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet b --dbpath b1 --logpath log.b1 --port 27101 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet b --dbpath b2 --logpath log.b2 --port 27102 --fork --logappend --smallfiles --oplogSize 50

mongod --shardsvr --replSet c --dbpath c0 --logpath log.c0 --port 27200 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet c --dbpath c1 --logpath log.c1 --port 27201 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet c --dbpath c2 --logpath log.c2 --port 27202 --fork --logappend --smallfiles --oplogSize 50

mongod --shardsvr --replSet d --dbpath d0 --logpath log.d0 --port 27300 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet d --dbpath d1 --logpath log.d1 --port 27301 --fork --logappend --smallfiles --oplogSize 50
mongod --shardsvr --replSet d --dbpath d2 --logpath log.d2 --port 27302 --fork --logappend --smallfiles --oplogSize 50


Run the mongs processes:

mongos --configdb localhost:26050, localhost:26051,localhost:26052 --fork --logappend --logpath log.mongos0
mongos --configdb localhost:26050, localhost:26051,localhost:26052 --fork --logappend --logpath log.mongos1 --port 26061
mongos --configdb localhost:26050, localhost:26051,localhost:26052 --fork --logappend --logpath log.mongos2 --port 26062
mongos --configdb localhost:26050, localhost:26051,localhost:26052 --fork --logappend --logpath log.mongos3 --port 26063

It is important that clients talk to the mongod default server on 27017 port not directly to the servers. Best practices:

• Run mongos on the standard mongodb tcp port 27017. It gives the possibility to connect to the cluester for the rest of the world.
• Do not run shard servers mongod's nor config server on that default port 27017.

How many mongos processes should you run (generally)?

One per config serverOne per mongod processOne per shardOneHowever many you want but usually much more than one

The Config Database

Commands:

>mongo
>use config
>show collections

chunks
databases
lockpings
locks
mongos
settings
shards
system.indexes
version

Which are true?

To check via the mongo shell what's in the config database of a MongoDB cluster, you must connect it directly to one of the config servers.Three config servers is typical in a MongoDB cluster with 1,000 total machines.All the config servers have exactly the same data.

Adding the Initial Shards

For each shard:

1. initiate the replica set connecting to the first shards mongod's
1. ps -A | grep mongod | grep 2700
2. Connect to mongo
1. mongo --port 27000
2. Configuration of replica set:
1. rs.initiate()
2. rs.add("localhost:27001")
3. rs.add("localhost:27002")
4. rs.conf()
3. Disconnect from replica set
2. "add" the shard to the cluster:  sh.addShard(<>)
1. Connect to mongos
1. mongo
2. sh.addShard("a/localhost:27000")
3. rs.status()

Enable Sharding for a Collection

1. turn on sharding for the database
2. sh.shardCollection(<collection>,{<field name> : 1 }

In MongoDB v2.6, by default (i.e., if shardCollection hasn't been invoked) collections are:

On the primary shard and not shardedNot sharded but homed on different shards at randomAutomatically sharded on _id

Working with a Sharded Cluster

In general we will want to use the sharded key inquiries whenever possible on a shard collection. If not, the query will be sent to all the mongod's waiting for an answer. If we have one thousand of shard servers, we will have an overhead. If we create a new index of the field to query, it will send to the shard server that have the document using the appropiate chunk.

Cardinality & Monotonic Shard Keys 

Thing to consider to select the shard key:

1. It usually is a field that involved the most of your queries. It is common in queries for the collection.
2. To make sure that the shard key has suffient cardinality: There is enough values that ve can spread the data out from the collection to a lot of different shards
3. To make sure that the shard key has good granularity to the shard ley values
4. consider compound of shard keys in order to get granularity
5. Is the key monotonically increasing ? BSON "2object id" fo this

Shard Key Selection Example

To choose shard key in "Orders collection". It depends on what kind of data will access

1. "_id": will be the best and efficiently option:  { "_id": 1 }
2. "company":  { "company" : 1 }
3. compounds key:
1. { company : 1 , "_id" : 1}
2. { company : 1 , "date" : 1}

Suppose we are building an email server system and have a collection:

messages
--------
{_id:<message_id>, mailbox_id:___, 
 sender_id:___, subject:___, date:___, 
 body:___, read:<bool>, ...
}

We have millions of users (email addresses / mailboxes) so want to shard this collection. Of the choices listed, what's a good shard key?

{_id:1}{mailbox_id:1}{sender_id:1}{sender_id:1,read:1}

Process & Machine Layout

We have:

1. shard servers (mongod --shardsvr)
2. config servers (mongod --configsvr)
3. mongos processes (mongos): possibilities:
1. put the mongos process in the client to access to all of clusters. In this case all the traffic between the client and the mongos process will remain inside the client.
2. run the mongos on every single server. The client would connect to any member to the cluster at random to the mongos that would be on that default mongo port 27017 and then, the mongos will talk to anything it needs to talk to in the whole cluster. Different client machines would connect to different mongos instances. 

If I have 3 shards, 3 member replica sets, 3 config servers, and 6 mongos processes running, how many machines might we use for that?

While we could use more if we wanted, 9 machines would be just fine.9 machines works but is too low for best practice usage

Bulk Inserts & Pre-splitting

Normally we do not need to do but there are some situations where it can be quite useful, in particulary:

1. sometimes with bulk inicial loads: a case can be when we need to pre-split the initial key range manually because of the loading data into the shard server one is faster than we are migrating it out and we are creating a backlog in this shard server 1.

Generally speaking, shard keys with non-uniform key distributions are:

OkOk only if you "pre-split"Not ok

Further Tips & Best Practices

1. Only shards the big collections
2. Pick shard key carefully: You cannot change the shardkey once it is set. You wuld have to create a new collection and copy the data over to it.
3. Considere pre-splitting some chunks on a bulk load
4. Be aware to monotically increasing shard key values on inserts which could happen if we are sharded on a timestamp field or BSON Object values. It can be just fine, but we want to be aware of the behaviour.
5. Adding new shards is easy but takes time because it's an online operation and it does not disrupt operations of cluster. However, once a shard is added, it's initially empty, so data has to move over to it. If eachs shard in the system has a few terabytes of data, it is going to take a while for data to migrate over to that, new shard gradually. We can not instantly and resources are required for that information to move over
6. Always connect to talk to the cluster through the mongos process, not directly to the mongod's except for dba work where we really want to talk to a certain server and do someting that is administratative. It is important no to accidentally connect to mongod's. In order to prevent that, we can put the mongos on the default port (27017)
7. Keep non-mongos precesses off  of 271027 port to avoid mistakes. 
8. Use logical config server names for config servers rather than direct ip addresses or even host names. Otherwise, it is really hard to change the config servers later or replace one, as every machine in the whole cluster has a memory of what the config servers are. If we do change config servers, read the documentation on how to do that, because if some members of the cluster think that the server is a config server, an yet it is not, it is no good.