Well, assuming that we have downloaded the MongoDB binaries from the download site, extracted it, and have the resulting bin directory in the operating system's path variable. (This is not mandatory, but it really becomes convenient after doing so.) The binaries can be downloaded from http://www.mongodb.org/downloads after selecting your host operating system.

If you see the following line on the console, you have successfully started the server:

[initandlisten] waiting for connections on port 27017

Starting a server can't get easier than this. Despite the simplicity in starting the server, there are a lot of configuration options that can be used to tune the behavior of the server on startup. Most of the default options are sensible and need not be changed. With the default values, the server should be listening to port 27017 for new connections, and the logs will be printed out to the standard output.

There are times where we would like to configure some options on server startup. In the Installing single node MongoDB recipe, we will use some more start up options.

If you have already seen and executed the Installing single node MongoDB recipe, you need not do anything different. If all these prerequisites are met, we are good for this recipe.

Ok, this wasn't too difficult and is similar to the previous recipe, but we have some additional command-line options this time around. MongoDB actually supports quite a few options at startup, and we will see a list of the most common and important ones in my opinion:

For an exhaustive list of options that are available, use the --help or -h option. This list of options is not exhaustive, and we will see some more coming up in later recipes as and when we need them. In the next recipe, we will see how to use a configuration file instead of the command-line arguments.

If you have already executed the Installing single node MongoDB recipe, you need not do anything different as all the prerequisites of this recipe are the same.

The /data/mongo/db directory for the database and /logs/ for the logs should be created and present on your filesystem with the appropriate permissions to write to it and perform the following steps:

All the command-line options that we discussed in the previous recipe, Starting a single node instance using command-line options, hold true. We are just providing them in a configuration file instead. If you have not visited the previous recipe, I would recommend you to do so as that is where we discussed some of the common command-line options. The properties are specified as <property name> = <value>. For all the properties that don't have values, for example, the smallfiles option, the value given is a Boolean value, true. If we need to have a verbose output, we would add v=true (or multiple v's to make it more verbose) to our configuration file. If you already know what the command-line option is, then it is pretty easy to guess what the value of the property is in the file. It is almost the same as the command-line option with just the hyphen removed.

Although it is possible to run the mongo shell without connecting to the MongoDB server using mongo --nodb, we would rarely need to do so. To start a server on the localhost without much of a hassle, take a look at the first recipe, Installing single node MongoDB, and start the server.

The JavaScript function that we executed here is of no practical use and is just used to demonstrate how a function can be preloaded on the startup of the shell. There could be multiple functions in the .js file containing valid JavaScript code—possibly some complex business logic.

On executing the mongo command without any arguments, we connect to the MongoDB server running on localhost and listen for new connections on the default port 27017. Generally speaking, the format of the command is as follows:

mongo <options> <db address> <.js files>

In cases where there are no arguments passed to the mongo executable, it is equivalent to the passing of the db address as localhost:27017/test.

Let's look at some example values of the db address command-line option and its interpretation:

Now, there are quite a few options available on the mongo client too. We will see a few of them in the following table:

The following are the prerequisites for this recipe:

These were quite a lot of steps to follow. Let's look at some of them in more detail. Everything up to step 6 is straightforward and doesn't need any explanation. Let's look at step 7 onwards.

The pom.xml file that we have here is pretty simple. We defined a dependency on mongo's Java driver. It relies on the online repository, repo.maven.apache.org, to resolve the artifacts. For a local repository, all we need to do is define the repositories and pluginRepositories tags in pom.xml. For more information on maven, refer to the maven documentation at http://maven.apache.org/guides/index.html.

For the Java class, the org.mongodb.MongoClient class is the backbone. We first instantiate it using one of its overloaded constructors giving the server's host and port. In this case, the hostname and port were not really needed as the values provided are the default values anyway, and the no-argument constructor would have worked well too. The following code snippet instantiates this client:

MongoClient client = new MongoClient("localhost", 27017);

The next step is to get the database, in this case, test using the getDB method. This is returned as an object of the com.mongodb.DB type. Note that this database might not exist, yet getDB will not throw any exception. Instead, the database will get created whenever we add a new document to the collection in this database. Similarly, getCollection on the DB object will return an object of the com.mongodb.DBCollection type representing the collection in the database. This too might not exist in the database and will get created on inserting the first document automatically.

The following two code snippets from our class show you how to get an instance of DB and DBCollection:

DB testDB = client.getDB("test");
DBCollection collection = testDB.getCollection("person");

Before we insert a document, we will drop the collection so that even upon multiple executions of the program, we will have just one document in the person collection. The collection is dropped using the drop() method on the DBCollection object's instance. Next, we create an instance of com.mongodb.DBObject. This is an object that represents the document to be inserted into the collection. The concrete class used here is BasicDBObject, which is a type of java.util.LinkedHashMap, where the key is String and the value is Object. The value can be another DBObject too, in which case, it is a document nested within another document. In our case, we have two keys, name and age, which are the field names in the document to be inserted and the values are of the String and Integer types, respectively. The append method of BasicDBObject adds a new key value pair to the BasicDBObject instance and returns the same instance, which allows us to chain the append method calls to add multiple key value pairs. This created DBObject is then inserted into the collection using the insert method. This is how we instantiated DBObject for the person collection and inserted it into the collection as follows:

DBObject person = new BasicDBObject("name", "Fred").append("age", 30);
collection.insert(person);

The findOne method on DBCollection is straightforward and returns one document from the collection. This version of findOne doesn't accept DBObject (which otherwise acts as a query executed before a document is selected and returned) as a parameter. This is synonymous to doing db.person.findOne() from the shell.

Finally, we simply invoke getDatabaseNames to get a list of databases' names in the server. At this point of time, we should at least be having test and the local database in the returned result. Once all the operations are complete, we close the client. The MongoClient class is thread-safe and generally one instance is used per application. To execute the program, we use the maven's exec plugin. On executing step 9, we should see the following lines toward the end in the console:

[INFO] [exec:java {execution: default-cli}]
--snip--
Dropping person collection in test database
Adding a person document in the person collection of test database
Now finding a person using findOne
Person found, name is Fred and age is 30
Database names are
1: local
2: test
INFO: Closed connection [connectionId{localValue:2, serverValue:2}] to localhost:27017 because the pool has been closed.
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESSFUL
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 3 seconds
[INFO] Finished at: Tue May 12 07:33:00 UTC 2015
[INFO] Final Memory: 22M/53M
[INFO] ------------------------------------------------------------------------ 

The following are the prerequisites for this recipe:

We start off by installing the Python MongoDB driver, pymongo, on the system with the help of the pip package manager. In the given Python code, we begin by importing print_function from the __future__ module to allow compatibility with Python 3.x. Next, we import pymongo so that it can be used in the script.

We instantiate pymongo.MongoClient() with localhost and 27017 as the mongo server host and port, respectively. In pymongo, we can directly refer to the database and its collection by using the <client>.<database_name>.<collection_name> convention.

In our recipe, we used the client handler to select the database test simply by referring to client.test. This returns a database object even if the database does not exist. As a part of this recipe, we drop the collection by calling testdb.person.drop(), where testdb is a reference to client.test and person is a collection that we wish to drop. For this recipe, we are intentionally dropping the collection so that recurring runs will always yield one record in the collection.

Next, we instantiate a dictionary called employee with a few values such as name and age. We will now add this entry to our person collection using the insert_one() method.

As we now know that there is an entry in the person collection, we will fetch one document using the find_one() method. This method returns the first document in the collection, depending on the order of documents stored on the disk.

Following this, we also try to get the list of all the databases by calling the get_databases() method to the client. This method returns a list of database names present on the server. This method may come in handy when you are trying to assert the existence of a database on the server.

Finally, we close the client connection using the close() method.

Though not a prerequisite, taking a look at the Starting a single node instance using command-line options recipe will definitely make things easier just in case you are not aware of various command-line options and their significance while starting a mongo server. Additionally, the necessary binaries and setups as mentioned in the single server setup must be done before we continue with this recipe. Let's sum up on what we need to do.

We will start three mongod processes (mongo server instances) on our localhost.

We will create three data directories, /data/n1, /data/n2, and /data/n3 for Node1, Node2, and Node3, respectively. Similarly, we will redirect the logs to /logs/n1.log, /logs/n2.log, and /logs/n3.log. The following image will give you an idea on how the cluster would look:

Let's take a look at the steps in detail:

We described the common options in the Installing single node MongoDB recipe with the command-line options recipe before and all these command-line options are described in detail.

As we are starting three independent mongod services, we have three dedicated database paths on the filesystem. Similarly, we have three separate log file locations for each of the processes. We then start three mongod processes with the database and log file path specified. As this setup is for test purposes and is started on the same machine, we use the --smallfiles and --oplogSize options. As these processes are running on the same host, we also choose the ports explicitly to avoid port conflicts. The ports that we chose here were 27000, 27001, and 27002. When we start the servers on different hosts, we may or may not choose a separate port. We can very well choose to use the default one whenever possible.

The --fork option demands some explanation. By choosing this option, we start the server as a background process from our operating system's shell and get the control back in the shell where we can then start more such mongod processes or perform other operations. In the absence of the --fork option, we cannot start more than one process per shell and would need to start three mongod processes in three separate shells.

If we take a look at the logs generated in the log directory, we should see the following lines in it:

[rsStart] replSet can't get local.system.replset config from self or any seed (EMPTYCONFIG)
[rsStart] replSet info you may need to run replSetInitiate -- rs.initiate() in the shell -- if that is not already done

Though we started three mongod processes with the --replSet option, we still haven't configured them to work with each other as a replica set. This command-line option is just used to tell the server on startup that this process will be running as a part of a replica set. The name of the replica set is the same as the value of this option passed on the command prompt. This also explains why the insert operation executed on one of the nodes failed before the replica set was initialized. In mongo replica sets, there can be only one primary node where all the inserting and querying happens. In the image shown, the N1 node is shown as the primary and listens to port 27000 for client connections. All the other nodes are slave/secondary instances, which sync themselves up with the primary and hence querying too is disabled on them by default. It is only when the primary goes down that one of the secondary takes over and becomes a primary node. However, it is possible to query the secondary for data as we have shown in the image; we will see how to query from a secondary instance in the next recipe.

Well, all that is left now is to configure the replica set by grouping the three processes that we started. This is done by first defining a JSON object as follows:

cfg = {
  '_id':'repSetTest', 'members':[ {'_id':0, 'host': 'localhost:27000'}, {'_id':1, 'host': 'localhost:27001'}, {'_id':2, 'host': 'localhost:27002'} ]
}

There are two fields, _id and members, for the unique ID of the replica set and an array of the hostnames and port numbers of the mongod server processes as part of this replica set, respectively. Using localhost to refer to the host is not a very good idea and is usually discouraged; however, in this case, as we started all the processes on the same machine, we are ok with it. It is preferred that you refer to the hosts by their hostnames even if they are running on localhost. Note that you cannot mix referring to the instances using localhost and hostnames both in the same configuration. It is either the hostname or localhost. To configure the replica set, we then connect to any one of the three running mongod processes; in this case, we connect to the first one and then execute the following from the shell:

> rs.initiate(cfg)

The _id field in the cfg object passed has a value that is the same as the value we gave to the --replSet option on the command prompt when we started the server processes. Not giving the same value would throw the following error:

{
        "ok" : 0,
        "errmsg" : "couldn't initiate : set name does not match the set name host Amol-PC:27000 expects"
}

If all goes well and the initiate call is successful, we should see something similar to the following JSON response on the shell:

{"ok" : 1}

In a few seconds, you should see a different prompt for the shell that we executed this command from. It should now become a primary or secondary. The following is an example of the shell connected to a primary member of the replica set:

repSetTest:PRIMARY>

Executing rs.status() should give us some stats on the replica set's status, which we will explore in depth in a recipe later in the book in the administration section. For now, the stateStr field is important and contains the PRIMARY, SECONDARY, and other texts.

Look at the Connecting to the replica set in the shell to query and insert data recipe to perform more operations from the shell after connecting to a replica set. Replication isn't as simple as we saw here. See the administration section for more advanced recipes on replication.

If you are looking to convert a standalone instance to a replica set, then the instance with the data needs to become a primary first, and then empty secondary instances will be added to which the data will be synchronized. Refer to the following URL on how to perform this operation:

http://docs.mongodb.org/manual/tutorial/convert-standalone-to-replica-set/

The prerequisite for this recipe is that the replica set should be set up and running. Refer to the previous recipe, Starting multiple instances as part of a replica set, for details on how to start the replica set.

We have done a lot of things in this recipe, and we will try to throw some light on some of the important concepts to remember.

We basically connect to a primary and secondary node from the shell and perform (I would say, try to perform) selects and inserts. The architecture of a Mongo replica set is made of one primary (just one, no more, no less) and multiple secondary nodes. All writes happen on the PRIMARY only. Note that replication is not a mechanism to distribute the read request load that enables scaling the system. Its primary intent is to ensure high availability of data. By default, we are not permitted to read data from the secondary nodes. In step 6, we simply insert data from the primary node and then execute a query to get the document that we inserted. This is straightforward and nothing related to clustering here. Just note that we inserted the document from the primary and then queried it back.

In the next step, we execute the same query but this time, from the secondary's shell. By default, querying is not enabled on the SECONDARY. There might be a small lag in replicating the data possibly due to heavy data volumes to be replicated, network latency, or hardware capacity to name a few of the causes, and thus, querying on the secondary might not reflect the latest inserts or updates made on the primary. However, if we are ok with it and can live with the slight lag in the data being replicated, all we need to do is enable querying on the SECONDARY node explicitly by just executing one command, rs.slaveOk() or rs.slaveOk(true). Once this is done, we are free to execute queries on the secondary nodes too.

Finally, we try to insert the data into a collection of the slave node. Under no circumstances is this permitted, regardless of whether we have done rs.slaveOk(). When rs.slaveOk() is invoked, it just permits the data to be queried from the SECONDARY node. All write operations still have to go to the primary and then flow down to the secondary. The internals of replication will be covered in a different recipe in the administration section.

The next recipe, Connecting to the replica set to query and insert data from a Java client, is about connecting to a replica set from a Java client.

We need to take a look at the Connecting to the single node using a Java client recipe as it contains all the prerequisites and steps to set up maven and other dependencies. As we are dealing with a Java client for replica sets, a replica set must be up and running. Refer to the Starting multiple instances as part of a replica set recipe for details on how to start the replica set.

An interesting thing to observe is how we instantiate the MongoClient instance. It is done as follows:

  MongoClient client = new MongoClient(Arrays.asList(new ServerAddress("localhost", 27000), new ServerAddress("localhost", 27001), new ServerAddress("localhost", 27002)));

The constructor takes a list of com.mongodb.ServerAddress. This class has a lot of overloaded constructors but we choose to use the one that takes the hostname and then port. What we have done is provided all the server details in a replica set as a list. We haven't mentioned what is the PRIMARY node and what are the SECONDARY nodes. MongoClient is intelligent enough to figure this out and connect to the appropriate instance. The list of servers provided is called the seed list. It need not contain an entire set of servers in a replica set though the objective is to provide as much as we can. MongoClient will figure out all the server details from the provided subset. For example, if the replica set is of five nodes but we provide only three servers, it works fine. On connecting with the provided replica set servers, the client will query them to get the replica set metadata and figure out the rest of the provided servers in the replica set. In the preceding case, we instantiated the client with three instances in the replica set. If the replica set was to have five members, then instantiating the client with just three of them is still good enough and the remaining two instances will be automatically discovered.

Next, we start the client from the command prompt using maven. Once the client is running in the loop, we bring down the primary instance to find one document. We should see something as the following output to the console:

_id: 1, value is abc
Now Retrieving documents in a loop from the collection.
Stop the primary instance manually after few iterations
_id: 1, value is abc
_id: 1, value is abc
Nov 03, 2013 5:21:57 PM com.mongodb.ConnectionStatus$UpdatableNode update
WARNING: Server seen down: Amol-PC/192.168.1.171:27002
java.net.SocketException: Software caused connection abort: recv failed
        at java.net.SocketInputStream.socketRead0(Native Method)
        at java.net.SocketInputStream.read(SocketInputStream.java:150)
        …
WARNING: Primary switching from Amol-PC/192.168.1.171:27002 to Amol-PC/192.168.1.171:27001
_id: 1, value is abc

As we can see, the query in the loop was interrupted when the primary node went down. However, the client switched to the new primary seamlessly. Well, nearly seamlessly, as the client might have to catch an exception and retry the operation after a predetermined interval has elapsed.

Refer to the Connecting to the single node using a Python client recipe as it describes how to set up and install PyMongo, the Python driver for MongoDB. Additionally, a replica set must be up and running. Refer to the Starting multiple instances as part of a replica set recipe for details on how to start the replica set.

You will notice that, in this script, we instantiated the mongo client by giving a list of hosts instead of a single host. As of version 3.0, the pymongo driver's MongoClient() class can accept either a list of hosts or a single host during initialization and deprecate MongoReplicaSetClient(). The client will attempt to connect to the first host in the list, and if successful, will be able to determine the other nodes in the replica set. We are also passing the replicaSet='repSetTest' parameter exclusively, ensuring that the client checks whether the connected node is a part of this replica set.

Once connected, we perform normal database operations such as selecting the test database, dropping the repTest collection, and inserting a single document into the collection.

Following this, we enter a conditional for loop, iterating five times. Each time, we fetch the record, display it, and sleep for three seconds. While the script is in this loop, we shut down the primary node in the replica set as mentioned in step 4. We should see an output similar to this:

Fetching record: {u'age': u'30', u'_id': ObjectId('5558bfaa0640fd1923fce1a1'), u'name': u'Foo'}
Fetching record: {u'age': u'30', u'_id': ObjectId('5558bfaa0640fd1923fce1a1'), u'name': u'Foo'}
Fetching record: {u'age': u'30', u'_id': ObjectId('5558bfaa0640fd1923fce1a1'), u'name': u'Foo'}
Could not connect to primary
Fetching record: {u'age': u'30', u'_id': ObjectId('5558bfaa0640fd1923fce1a1'), u'name': u'Foo'}

In the preceding output, the client gets disconnected from the primary node midway. However, very soon, a new primary node is selected by the remaining nodes and the mongo client is able to resume the connection.

Apart from the MongoDB server already installed, no prerequisites are there from a software perspective. We will be creating two data directories, one for each shard. There will be a directory for the data and one for logs.

Let's see what all we did in the process. We created three directories for data (two for the shards and one for the configuration database) and one directory for logs. We can have a shell script or batch file to create the directories as well. In fact, in large production deployments, setting up shards manually is not only time-consuming but also error-prone.

Let's try to get a picture of what exactly we have done and are trying to achieve. The following is an image of the shard setup that we just did:

If we look at the preceding image and the servers started in step 2, we have shard servers that would store the actual data in the collections. These were the first two of the four processes that we started listening to ports 27000 and 27001. Next, we started a configuration server that is seen on the left side in this image. It is the third server of the four servers started in step 2 and it listens to port 25000 for the incoming connections. The sole purpose of this database is to maintain the metadata about the shard servers. Ideally, only the mongos process or drivers connect to this server for the shard details/metadata and the shard key information. We will see what a shard key is in the next recipe, where we play around a sharded collection and see the shards that we have created in action.

Finally, we have a mongos process. This is a lightweight process that doesn't do any persistence of data and just accepts connections from clients. This is the layer that acts as a gatekeeper and abstracts the client from the concept of shards. For now, we can view it as basically a router that consults the configuration server and takes the decision to route the client's query to the appropriate shard server for execution. It then aggregates the result from various shards if applicable and returns the result to the client. It is safe to say that no client connects directly to the configuration or shard servers; in fact, no one ideally should connect to these processes directly except for some administration operations. Clients simply connect to the mongos process and execute their queries and insert or update operations.

Just starting the shard server, configuration server, and mongos process doesn't create a sharded environment. On starting up the mongos process, we provided it with the details of the configuration server. What about the two shards that would be storing the actual data? However, the two mongod processes started as shard servers are not yet declared anywhere as shard servers in the configuration. This is exactly what we do in the final step by invoking sh.addShard() for both the shard servers. The mongos process is provided with the configuration server's details on startup. Adding shards from the shell stores this metadata about the shards in the configuration database, and the mongos processes then would be querying this config database for the shard's information. On executing all the steps of the recipe, we have an operational shard as follows:

Before we conclude, the shard that we have set up here is far from ideal and not how it would be done in a production environment. The preceding image gives us an idea of how a typical shard would be in a production environment. The number of shards would not be two but many more. Additionally, each shard will be a replica set to ensure high availability. There would be three configuration servers to ensure availability of the configuration servers as well. Similarly, there will be any number of mongos processes created for a shard listening for client connections. In some cases, it might even be started on a client application's server.

What good is a shard unless we put it to action and see what happens from the shell on inserting and querying the data? In the next recipe, we will make use of the shard setup here, add some data, and see it in action.

Obviously, we need a sharded mongo server setup up and running. See the previous recipe, Starting a simple sharded environment of two shards, for more details on how to set up a simple shard. The mongos process, as in the previous recipe, should be listening to port number 27017. We have got some names in a JavaScript file called names.js. This file needs to be downloaded from the Packt website and kept on the local filesystem. The file contains a variable called names and the value is an array with some JSON documents as the values, each one representing a person. The contents look as follows:

names = [
  {name:'James Smith', age:30},
  {name:'Robert Johnson', age:22},
…
]

This recipe demands some explanation. We downloaded a JavaScript file that defines an array of 20 people. Each element of the array is a JSON object with the name and age attributes. We start the shell connecting to the mongos process loaded with this JavaScript file. We then switch to shardDB, which we use for the purpose of sharding.

For a collection to be sharded, the database in which it will be created needs to be enabled for the sharding first. We do this using sh.enableSharding().

The next step is to enable the collection to be sharded. By default, all the data will be kept on one shard and not split across different shards. Think about it; how will Mongo be able to split the data meaningfully? The whole intention is to split it meaningfully and as evenly as possible so that whenever we query based on the shard key, Mongo would easily be able to determine which shard(s) to query. If a query doesn't contain the shard key, the execution of the query will happen on all the shards and the data would then be collated by the mongos process before returning it to the client. Thus, choosing the right shard key is very crucial.

Let's now see how to shard the collection. We do this by invoking sh.shardCollection("shardDB.person", {name: "hashed"}, false). There are three parameters here:

The last step is to see the execution plan for the finding of all the data. The intent of this operation is to see how the data is being split across two shards. With 300,000 documents, we expect something around 150,000 documents on each shard. However, from the distribution statistics, we can observe that shard0000 has 1,49,715 documents whereas shard0001 has 150285:

Shard shard0000 at localhost:27000
 data : 15.99MiB docs : 149715 chunks : 2
 estimated data per chunk : 7.99MiB
 estimated docs per chunk : 74857

Shard shard0001 at localhost:27001
 data : 16.05MiB docs : 150285 chunks : 2
 estimated data per chunk : 8.02MiB
 estimated docs per chunk : 75142

Totals
 data : 32.04MiB docs : 300000 chunks : 4
 Shard shard0000 contains 49.9% data, 49.9% docs in cluster, avg obj size on shard : 112B
 Shard shard0001 contains 50.09% data, 50.09% docs in cluster, avg obj size on shard : 112B

There are a couple of additional suggestions that I would recommend you to do.

Connect to the individual shard from the mongo shell and execute queries on the person collection. See that the counts in these collections are similar to what we see in the preceding plan. Additionally, one can find out that no document exists on both the shards at the same time.

We discussed in brief about how cardinality affects the way the data is split across shards. Let's do a simple exercise. We first drop the person collection and execute the shardCollection operation again but, this time, with the {name: 1} shard key instead of {name: "hashed"}. This ensures that the shard key is not hashed and stored as is. Now, load the data using the JavaScript function we used earlier in step number 5, and then execute the explain() command on the collection once the data is loaded. Observe how the data is now split (or not) across the shards.

A lot of questions must now be coming up such as what are the best practices? What are some tips and tricks? How is the sharding thing pulled off by MongoDB behind the scenes in a way that is transparent to the end user?

This recipe here only explained the basics. In the administration section, all such questions will be answered.