MongoDB from theory to practice. MongoDB Cluster Installation Guide

Good day, dear readers. In this post I would like to describe several examples of mongoDB scan, the differences between them, the principles of their work. However, most of all I would like to share with you the practical experience of sharding mongoDB. If this post had a plan, it would most likely look like this:

1. Introduction. Scaling brief
2. Some examples of mongoDB scan and their description
3. Sharding mongoDB

Points 1 and 2 are theoretical, and number 3 claims to be a practical guide to raising the mongoDB cluster and is best suited to those who are faced with this for the first time.

1. Entry Little about scaling

Imagine a typical case - there is a database in which data is written and read. In dynamically growing systems, data volumes tend to increase rapidly and sooner or later you may encounter a problem when the current resources of the machine are not enough for normal operation.
Scaling is used to solve this problem. Scaling happens 2 types - horizontal and vertical. Vertical scaling - increasing the capacity of one machine - adding CPU, RAM, HDD. Horizontal scaling - adding new machines to existing ones and distributing data between them. The first case is the simplest, because does not require additional settings of the application and any additional configuration of the database, but its disadvantage is that the power of one machine theoretically sooner or later come to a standstill. The second case is more complicated in configuration, but has several advantages:

• Theoretically infinite scaling (you can put as many machines as you like)
• Greater data security (only when using replication) - the machines can be located in different data centers (if one of them falls, others will remain)

2. Some examples of mongoDB scan

')

2.1 The simplest scheme, without sharding

In this scheme, everything is simple - there is an application that communicates with mongod through a driver. mongod is the main mongoDB process whose task is to receive requests, process and execute them. The data in mongod is stored in the so-called chunks. Each chunk has a size of “chunksize”, which is 64 MB by default. Physically, chunks are stored in dbName.n files, where n is a sequence number, starting from 0. When the size reaches 64 MB (or another chunksize) the chunk is divided in half, it turns out 2 less chanics - 32 MB each, these 2 chanks begin to fill until reach the size of chunksize, then the separation occurs again, etc. The dbName.0 file size is 64 MB, dbName.1 is 128 MB, dbName.2 is 256, and so on. up to 2Gb. As the number and size of chunks grows, these files are filled and when there is a first file - dbName.5, the size of which is 2 Gb, the size growth stops and mongoDB simply creates files of the same size. It should also be noted that mongoDB does not just create these files as necessary, but creates them in advance so that if it is necessary to actually write data to the file, do not waste time creating the file. Therefore, with a relatively small amount of real data, you can find that the hard disk space is decently occupied.
This scheme is applicable most often for local testing.

2.2 Shardirovannaya scheme without replica sets

In this scheme, new items appear. The pattern is called shardirovannoy. An important difference between the shardirovannogo scheme - her data is not just written in chunks, which are then divided in half, and fall into them on a certain range of a given field - shard key. First, only one chunk is created and the range of values ​​it takes is within (-∞, + ∞):



When the size of this chunk reaches chunksize, mongos evaluates the value of all the shardkei inside the chunk and divides the chunk in such a way that the data are divided approximately equally. For example, suppose that we have 4 documents with the fields name, age and id. id is the shard key:

{“name”: “Max”, “age”: 23, “id”: 23} {“name”: “John”, “age”: 28, “id”: 15} {“name”: “Nick”, “age”: 19, “id”:56} {“name”: “Carl”, “age”: 19, “id”: 78} 

Assume the size of chunksize, already reached. In this case, Mongos will divide the range like this (-, 45]; (45, +). We will have 2 chunk:



When new documents appear, they will be recorded in a chunk that corresponds to the shardKey range. Upon achieving chunksize, the separation will occur again and the range will be even narrower and so on. All chunks are stored on shards.
It is important to note that when a chunk reaches an indivisible range, for example (44, 45), the division will not occur and the chunk will grow above chunksize. Therefore, you should carefully select the shard key so that it is the most random variable possible. we need to fill in the database with all the people on the planet, then the successful shard key choices would be the phone number, the tax identification number, the postal code, the name, the city of residence were unsuccessful.
In the diagram we can see the config server, its difference from mongod is that it does not process client requests, but is a metadata repository - it knows the physical addresses of all the chunk, knows what chunk, what shard to look for and what range a particular another shard. It stores all this data in a specially designated place - config database.
This scheme also presents a query router, mongos, with the following tasks assigned to it:

1. Caching data stored on the config server
2. Routing of read and write requests from the driver - routing requests from applications to the necessary shards, mongos knows exactly where this or that chunk is physically located
3. Running the background process “Balancer”

The balancer function is to migrate chunks from one shard to another. The process goes something like this: the balancer sends the moveChunk command to the shard, from which the chunk will migrate, the shard receiving this command starts the process of copying the chunk to another shard. After all documents are copied, documents are synchronized between these 2 chunks, since while the migration was taking place, new data could be added to the source chunk. After the end of synchronization, the shard, which received the new chunk, sends its config address to the server so that it, in turn, updates it in the Mongos cache. At the end of this process, if there are no open cursors on the original chunk, it is deleted.
This scheme often takes place in a test environment of large applications, and when using 3-config servers, it may be suitable for small production applications. 3 server configs provide data redundancy and if one falls, mongos will still receive the actual addresses of the chunks from other server config.

2.3 Shardirovannaya scheme with replica sets

In this scheme, in addition to sharding, there is replication of shards. A few words about it. All writes, deletions, updates, get into the master (primary), and then are recorded in a special collection of oplog, from where they asynchronously get on replicas - repl.1 and repl.2 (secondary). Thus, data duplication occurs. Why do you need it?

• Redundancy ensures data security - when the master crashes, voting takes place between the replicas and one of them becomes the master.
• Master and replicas can be located in different data centers - this can be useful if the server is physically damaged (fire in the data center)
• Replicas can be used to more efficiently read data. For example, there is an application that has a client audience in Europe and in the USA. One of the replicas can be placed in the United States and set up so that customers from the United States read data from it. It is worth noting that the documents for the replicas are delayed and it is not always possible to immediately find a newly recorded document on the replica. Therefore, this item is an advantage only if the reading from the replicas allows the application logic

The scheme with a replica sets most often takes place in serious production applications, where data integrity is important or there are a large number of readings and the application logic allows reading from replicas.
We will not dwell on this scheme in more detail, since she can devote a separate post.

3. Sharding

So let's get started. We will expand all this locally on linux ubuntu 12.0. For all this we need the installed mongoDB, I have version 2.4.9.
We will shard the scheme number 2, just remove from it elements that have a purely theoretical value:



Let's open the familiar shell to all of us , preferably several tabs at once, since mongos, mongod, config server are all separate processes. Further points:

1. Create 2 empty directories in which data will be stored:
   

    > sudo mkdir /data/instance1 > sudo mkdir /data/instance2 

   
   Raise 2 mongod instances with commands:
   

    > sudo mongod --dbpath /data/instance1 --port 27000 //   

   
   And in the following terminal:
   

    > sudo mongod --dbpath /data/instance2 --port 27001 //  

   
   The --dbpath parameter specifies the path where the .0, .1, .2, and .ns files will be stored. In the files .0, .1, .2, etc. the data of this instance is stored in binary form, and in the .ns file the namespace required for navigating the database. --port - the port on which the database object will be available.
   After the first item, we have two instances:
   
   
2. Create an empty directory in which the server config data will be stored:
   

    > sudo mkdir /data/config 

   
   Raise the server configuration command
   

    > sudo mongod --configsvr --dbpath /data/config --port 27002 

   
   The --configsvr parameter indicates that the new instance will be exactly the config server, --dbpath is the path by which data will be stored. After the second point, the picture looks like this (let me draw your attention to the fact that while these entities know nothing about each other):
   
   
3. Raise mongos, team
   

    > sudo mongos --configdb 127.0.0.1:27002 --port 27100 

   
   By this command, mongos is raised on port 27100 , he needs to send a list of servers with their hosts to which he will access. If we did not specify a port when raising Mongos, then it uses the default 27017 (if it is not busy). After raising Mongos:
   
   
4. Connecting to Mongos, indicating the port on which we raised it, the team
   

    > mongo --port 27100 

   
   After that we get:
   
   
5. There is a final step - add our shards to the cluster
   

    > sh.addShard("127.0.0.1:27000") > sh.addShard("127.0.0.1:27001") 

   
   These 2 commands need to be executed on Mongos, the connection to which was opened in clause 4. Using the db.printShardingStatus () command, you can view the status of sharding. To make sure the shards are added, in the terminal we should see something like:
   
   

    --- Sharding Status --- sharding version: { "_id" : 1, "version" : 3, "minCompatibleVersion" : 3, "currentVersion" : 4, "clusterId" : ObjectId("53317aefca1ba9ba232b949e") } shards: { "_id" : "shard0000", "host" : "127.0.0.1:27000" } { "_id" : "shard0001", "host" : "127.0.0.1:27001" } databases: { "_id" : "admin", "partitioned" : false, "primary" : "config" } { "_id" : "test", "partitioned" : false, "primary" : "shard0000" } 

   
   
   We have the final picture:
   
   

Now let's make sure that everything we set up works, namely, the data is formed into chunks, and the balancer scatters them around the shards. In order for the data to begin to be shaded, you must enable sharding on the required database, and then on the collection.
All of the following commands must be performed under Mongos. Let our database be called a bank , we will execute commands that allow it to be shaded:

 > use admin > sh.enableSharding("bank") 

Run the db.printShardingStatus () command again . The output should be something like this:

 databases: { "_id" : "admin", "partitioned" : false, "primary" : "config" } { "_id" : "test", "partitioned" : false, "primary" : "shard0000" } { "_id" : "bank", "partitioned" : true, "primary" : "shard0000" } 

As you can see, in front of partitioned , true appeared, which means we're on the right track.

Now let's work with our database:

 > use bank //   bank (  ,   ) > db.tickets.insert({name: “Max”, amount: Math.random()*100}) //     tickets (  ,      ) > for (var i=0; i<2100000; i++) { db.tickets.insert({name: “Max”, amount: Math.random()*100}) } //     javascript > db.tickets.ensureIndex({amount: 1}) // ,       shard Key > db.tickets.stats() //,    .  ,   sharded   false,       primaryShard > use admin //  ,     tickets.         (primary)  > db.runCommand({shardCollection: "bank.tickets", key: {amount: 1}}) //  ,    - shard key  . 

After the last command, we should see something like this:

 { "collectionsharded" : "bank.tickets", "ok" : 1 } 

After all this, run the sh.status (true) or db.printShardingStatus () command to make sure that everything worked and, if everything is done correctly, we should see the following picture:



As you can see, the data is unevenly distributed, but if you wait a bit and repeat the db.printShardingStatus () command, the picture changes in the direction of a uniform distribution:





And the final picture:



As we have seen, at first chunks are saved on the primary shard, and then migrate to the second shard until the number evens out, and their ranges can also change.

In the future I would like to tell you about the organization of memory in mongoDB and about replication. Thanks for attention.