- Sharding >
- Sharding Concepts >
- Sharded Cluster Behavior >
- Sharded Cluster Query Routing
Sharded Cluster Query Routing¶
On this page
MongoDB mongos instances route queries and write operations
to shards in a sharded cluster. mongos provide the
only interface to a sharded cluster from the perspective of
applications. Applications never connect or communicate directly with
the shards.
The mongos tracks what data is on which shard by caching
the metadata from the config servers. The mongos uses the
metadata to route operations from applications and clients to the
mongod instances. A mongos has no persistent
state and consumes minimal system resources.
The most common practice is to run mongos instances on the
same systems as your application servers, but you can maintain
mongos instances on the shards or on other dedicated
resources.
Note
Changed in version 2.1.
Some aggregation operations using the aggregate
command (i.e. db.collection.aggregate()) will cause
mongos instances to require more CPU resources than in
previous versions. This modified performance profile may dictate
alternate architecture decisions if you use the aggregation
framework extensively in a sharded environment.
Routing Process¶
A mongos instance uses the following processes to route
queries and return results.
How mongos Determines which Shards Receive a Query¶
A mongos instance routes a query to a cluster by:
- Determining the list of shards that must receive the query.
- Establishing a cursor on all targeted shards.
In some cases, when the shard key or a prefix of the shard
key is a part of the query, the mongos can route the
query to a subset of the shards. Otherwise, the mongos
must direct the query to all shards that hold documents for that
collection.
Example
Given the following shard key:
Depending on the distribution of chunks in the cluster, the
mongos may be able to target the query at a subset of
shards, if the query contains the following fields:
How mongos Handles Query Modifiers¶
If the result of the query is not sorted, the mongos
instance opens a result cursor that “round robins” results from all
cursors on the shards.
Changed in version 2.0.5: In versions prior to 2.0.5, the mongos exhausted each
cursor, one by one.
If the query specifies sorted results using the
sort() cursor method, the mongos instance
passes the $orderby option to the shards. When the
mongos receives results it performs an incremental merge sort
of the results while returning them to the client.
If the query limits the size of the result set using the
limit() cursor method, the mongos
instance passes that limit to the shards and then re-applies the limit
to the result before returning the result to the client.
If the query specifies a number of records to skip using the
skip() cursor method, the mongos cannot
pass the skip to the shards, but rather retrieves unskipped results
from the shards and skips the appropriate number of documents when assembling
the complete result. However, when used in conjunction with a
limit(), the mongos will pass the limit
plus the value of the skip() to the shards to
improve the efficiency of these operations.
Detect Connections to mongos Instances¶
To detect if the MongoDB instance that your client is connected
to is mongos, use the isMaster command. When a
client connects to a mongos, isMaster returns
a document with a msg field that holds the string
isdbgrid. For example:
If the application is instead connected to a mongod, the
returned document does not include the isdbgrid string.
Broadcast Operations and Targeted Operations¶
In general, operations in a sharded environment are either:
- Broadcast to all shards in the cluster that hold documents in a collection
- Targeted at a single shard or a limited group of shards, based on the shard key
For best performance, use targeted operations whenever possible. While some operations must broadcast to all shards, you can ensure MongoDB uses targeted operations whenever possible by always including the shard key.
Broadcast Operations¶
mongos instances broadcast queries to all shards for the
collection unless the mongos can
determine which shard or subset of shards stores this data.
Multi-update operations are always broadcast operations.
The remove() operation is always a
broadcast operation, unless the operation specifies the shard key in
full.
Targeted Operations¶
All insert() operations target to one
shard.
All single update() (including upsert
operations) and remove() operations must
target to one shard.
Important
All single update() and
remove() operations must include the
shard key or the _id field in the query
specification. update() or
remove() operations that affect a single
document in a sharded collection without the shard key or
the _id field return an error.
For queries that include the shard key or portion of the shard key,
mongos can target the query at a specific shard or set of
shards. This is the case only if the portion of the shard key included
in the query is a prefix of the shard key. For example, if the shard
key is:
The mongos program can route queries that include the full
shard key or either of the following shard key prefixes at a
specific shard or set of shards:
Depending on the distribution of data in the cluster and the
selectivity of the query, mongos may still have to
contact multiple shards [1] to fulfill these queries.
| [1] | mongos will route some queries, even
some that include the shard key, to all shards, if needed. |
Sharded and Non-Sharded Data¶
Sharding operates on the collection level. You can shard multiple collections within a database or have multiple databases with sharding enabled. [2] However, in production deployments, some databases and collections will use sharding, while other databases and collections will only reside on a single shard.
Regardless of the data architecture of your sharded cluster,
ensure that all queries and operations use the mongos router to
access the data cluster. Use the mongos even for operations
that do not impact the sharded data.
| [2] | As you configure sharding, you will use the
enableSharding command to enable sharding for a
database. This simply makes it possible to use the
shardCollection command on a collection within that database. |