7.5.6 Controlling Collection Distribution (Tag-Based Sharding)
In the previous section, you saw how data
distribution happens. In this section, you will learn about
tag-based sharding . This feature was introduced in version 2.2.0.
In order to understand tag-based sharding,
let’s set up a sharded cluster. You will be using the shard
cluster created above. For this example, you need three shards, so
you will add Shard2 again to the cluster.
7.5.6.1 Prerequisite
1.
2.
3.
Since you have removed Shard2 from the
sharded cluster in the earlier example, you must add Shard2 to the
sharded cluster because for this example you need three shards.
If you try adding the removed shard without
removing the testdb database, it will give the following error:
7.5.6.2 Tagging
By the end of the above steps you have your
sharded cluster with a config server, three shards, and a mongos up
and running. Next, connect to the mongos at 30999 port and configdb
at 27022 in a new terminal window:
2015-07-13T23:24:39.674-0700 W SHARDING
running with 1 config server should be done only for testing
purposes and is not recommended for production
2015-07-13T23:24:39.931-0700 I SHARDING
[Balancer] distributed lock 'balancer /ANOC9:30999: 1429851279:41'
unlocked..
1.
2.
3.
4.
5.
6.
7.
mongos>for(var
i=0;i<100000;i++){db.drama.insert({originality:Math.random(),
count:i, time:new Date()});}
mongos>for(var i=0;i<100000;i++)
{db.comedy.insert({collections:Math.random(), count:i, time:new
Date()});}
By the end of the above step you have
three shards and three collections with sharding enabled on the
collections. Next you will see how data is distributed across the
shards.
{ "shard" : "shard0001" }
As you can see, the chunks are pretty
evenly spread out amongst the shards. See Figure 7-22.
Figure 7-22.
Distribution without tagging
Next, you will use tags to separate the
collections. The intent of this is to have one collection per shard
(i.e. your goal is to have the chunk distribution shown in Table
7-6).
Table 7-6.
Chunk Distribution
|
Collection Chunks |
On Shard |
|---|---|
|
movies.drama |
Shard0000 |
|
movies.action |
Shard0001 |
|
movies.comedy |
Shard0002 |
A tag describes the shard’s property,
which can be anything. Hence you might tag a shard as “slow” or
“fast” or “rack space” or “west coast.”
The rule uses MinKey, which means negative
infinity, and MaxKey, which means positive infinity. Hence the
above rule means mark all of the chunks of the collection
movies.drama with the tag “dramas.”
You need to wait for the cluster to
rebalance so that the chunks are distributed based on the tags and
rules defined above. As mentioned, the chunk distribution is an
automatic process, so after some time the chunks will automatically
be redistributed to implement the changes you have made.
Thus the collection chunks have been
redistributed based on the tags and rules defined (Figure 7-23).
Figure 7-23.
Distribution with tagging
7.5.6.3 Scaling with Tagging
Next, you will look at how to scale with
tagging . Let’s change the scenario. Let’s assume the
collection movies.action needs two servers for its data. Since you
have only three shards, this means the other two collection’s
data need to be moved to one shard.
In this scenario, you will change the
tagging of the shards. You will add the tag “comedies” to
Shard0 and remove the tag from Shard2, and further add the tag
“actions” to Shard2.
This means that the chunks tagged
“comedies” will be moved to Shard0 and chunks tagged “actions”
will be spread to Shard2.
Next, you add the tag “actions” to
Shard2, so that movies.action chunks are spread across Shard2 also:
The chunks have been redistributed
reflecting the changes made (Figure 7-24).
Figure 7-24.
Tagging with scaling
7.5.6.4 Multiple Tags
Say you want to distribute the writes
based on the disk. You have one shard that has a spinning disk and
the other has a SSD (solid state drive). You want to redirect 50%
of the writes to the shard with SSD and the remaining to the one
with the spinning disk.
Let’s further assume you have a
distribution field of the movies.action collection that you will be
using as the shard key. The distribution field value is between 0
and 1. Next, you want to say, “If distribution < .5, send this
to the spinning disk. If distribution >= .5, send to the SSD.”
So you define the rules as follows:
7.5.7 Points to Remember When Importing Data in a ShardedEnvironment
7.5.7.1 Pre-Splitting of the Data
7.5.7.2 Deciding on the Chunk Size
1.
2.
7.5.8 Monitoring for Sharding
In addition to the normal monitoring and
analysis that is done for other MongoDB instances, the sharding
cluster requires an additional monitoring to ensure that all its
operations are functioning appropriately and the data is
distributed effectively among the nodes. In this section, you will
see what monitoring you should do for the proper functioning of the
sharding cluster.
7.5.9 Monitoring the Config Servers
The config server, as you know by now,
stores the metadata of the sharded cluster. The mongos caches the
data and routes the request to the respective shards. If the config
server goes down but there’s a running mongos instance, there’s
no immediate impact on the shard cluster and it will remain
available for a while. However, you won’t be able to perform
operations like chunk migration or restart a new mongos. In the
long run, the unavailability of the config server can severely
impact the availability of the cluster. To ensure that the cluster
remains balanced and available, you should monitor the config
servers.
7.5.9.1 Monitoring the Shard Status Balancing and Chunk Distribution
For a most effective sharded cluster
deployment , it’s required that the chunks be distributed evenly
among the shards. As you know by now, this is done automatically
by MongoDB using a background process. You need to monitor the
shard status to ensure that the process is working effectively.
For this, you can use the db.printShardingStatus() or sh.status()
command in the mongos mongo shell to ensure that the process is
working effectively.
7.5.9.2 Monitoring the Lock Status
In almost all cases the balancer releases
its locks automatically after completing its process, but you need
to check the lock status of the database in order to ensure
there’s no long lasting lock because this can block future
balancing, which will affect the availability of the cluster.
Issue the following from mongos mongo to check the lock status:
7.6 Production Cluster Architecture
In this section, you will look at the
production cluster architecture. In order to understand it, let’s
consider a very generic use case of a social networking application
where the user can create a circle of friends and can share their
comments or pictures across the group. The user can also comment or
like her friend’s comments or pictures. The users are
geographically distributed.
The application
requirement is immediate availability across geographies of all the
comments; data should be redundant so that the user’s comments,
posts and pictures are not lost; and it should be highly available.
So the application’s production cluster should have the following
components :
1.
At least two
mongos
instance, but you can have more as per need.
2.
Three config
servers, each on a separate system.
3.
Two or more
replica
sets serving as shards.
The replica sets are distributed across geographies with read
concern set to nearest. See Figure 7-25.
Figure 7-25.
Production cluster architecture
Now let’s look at the possible failure
scenarios in MongoDB production deployment and its impact on the
environment.
7.6.1 Scenario 1
Mongos become unavailable: The
application server where mongos has gone down will not be able to
communicate with the cluster but it will not lead to any data loss
since the mongos don’t maintain any data of its own. The mongos
can restart, and while restarting, it can sync up with the config
servers to cache the cluster metadata, and the application can
normally start its operations (Figure 7-26).
Figure 7-26.
Mongos become unavailable
7.6.2 Scenario 2
One of the mongod of the replica set
becomes unavailable in a shard: Since you used replica sets to
provide high availability, there is no data loss. If a primary
node is down, a new primary is chosen, whereas if it’s a
secondary node, then it is disconnected and the functioning
continues normally (Figure 7-27).
Figure 7-27.
One of the mongod of replica set is
unavailable
The only difference is that the
duplication of the data is reduced, making the system little weak,
so you should in parallel check if the mongod is recoverable. If
it is, it should be recovered and restarted whereas if it’s
non-recoverable, you need to create a new replica set and replace
it as soon as possible.
7.6.3 Scenario 3
If one of the shard becomes unavailable:
In this scenario, the data on the shard will be unavailable, but
the other shards will be available, so it won’t stop the
application. The application can continue with its read/write
operations; however, the partial results must be dealt with within
the application. In parallel, the shard should attempt to recover
as soon as possible (Figure 7-28).
Figure 7-28.
Shard unavailable
7.6.4 Scenario 4
Only one config server is available out
of three: In this scenario, although the cluster will become
read-only, it will not serve any operations that might lead to
changes in the cluster structure, thereby leading to a change of
metadata such as chunk migration or chunk splitting. The config
servers should be replaced ASAP because if all config servers
become unavailable, this will lead to an inoperable cluster
(Figure 7-29).
Figure 7-29.
Only one config server available
No comments:
Post a Comment