Day 3: Replica Sets, Sharding, GeoSpatial, and GridFS

Mongo has a powerful ability to store and query data in a variety of ways. But then again, other databases can do those things, too. What makes document databases unique is their ability to efficiently handle arbitrarily nested, schemaless data documents. Thus far, we’ve run Mongo as a single server. But if you were to run Mongo in production, you’d want to run it as a cluster of machines, which would provide for much higher availability and enable you to replicate data across servers, shard collections into many pieces, and perform queries in parallel.

Replica Sets

Mongo was meant to scale out, not to run in standalone mode. It was built for data consistency and partition tolerance, but sharding data has a cost: If one part of a collection is lost, the whole thing is compromised. What good is querying against a collection of countries that contains only the Western Hemisphere or only Asia? Mongo deals with this implicit sharding weakness in a simple manner: duplication. You should rarely run a single Mongo instance in production and instead replicate the stored data across multiple services.

Rather than muck with our existing database, today we’ll start from scratch and spawn a few new servers. Mongo’s default port is 27017, so we’ll start up each server on other ports. Remember that you must create the data directories first, so create three of them:

Next, we’ll fire up the Mongo servers. This time we’ll add the replSet flag with the name book and specify the ports.

Open another terminal window and run the next command, which launches another server, pointing to a different directory, available on another port. Then open a third terminal to start the third server.

Notice that you get lots of this noise on the output, with error messages like this:

That’s a good thing because we haven’t yet initialized our replica set and Mongo is letting us know that. Fire up a mongo shell to one of the servers, and execute the rs.initiate function.

Notice we’re using a new object called rs (replica set) instead of db (database). Like other objects, it has a help method you can call. Running the status command will let us know when our replica set is running, so just keep checking the status for completion before continuing. If you watch the three server outputs, you should see that one server outputs this line:

And two servers will have the following output:

PRIMARY will be the master server. Chances are, this will be the server on port 27011 (because it started first); however, if it’s not, go ahead and fire up a console to the primary. Just insert any old thing on the command line, and we’ll try an experiment.

After the insert, exit the console, and then let’s test that our change has been replicated by shutting down the master node; pressing Ctrl+C is sufficient. If you watch the logs of the remaining two servers, you should see that one of the two has now been promoted to master (it will output the Member ... is now in state PRIMARY line). Open a console into that machine (for us it was localhost:27012), and db.echo.find() should contain your value.

We’ll play one more round of our console-shuffle game. Open a console into the remaining SECONDARY server by running mongo localhost:27013. Just to be sure, run the isMaster function. Ours looked like this:

In this shell, let’s attempt to insert another value.

This message is letting us know that we can neither write to a secondary node nor read directly from it. There is only one master per replica set, and you must interact with it. It is the gatekeeper to the set.

Replicating data has its own issues not found in single-source databases. In the Mongo setup, one problem is deciding who gets promoted when a master node goes down. Mongo deals with this by giving each mongod service a vote, and the one with the freshest data is elected the new master. Right now, you should still have two mongod services running. Go ahead and shut down the current master (aka primary node). Remember, when we did this with three nodes, one of the others just got promoted to be the new master. Now the last remaining node is implicitly the master.

Go ahead and relaunch the other servers and watch the logs. When the nodes are brought back up, they go into a recovery state and attempt to resync their data with the new master node. “Wait a minute!?” we hear you cry. “So, what if the original master had data that did not yet propagate?” Those operations are dropped. A write in a Mongo replica set isn’t considered successful until most nodes have a copy of the data.

The Problem with Even Nodes

The concept of replication is easy enough to grasp: You write to one MongoDB server, and that data is duplicated across others within the replica set. If one server is unavailable, then one of the others can be promoted and serve requests. But a server can be unavailable in more ways than a server crash. Sometimes, the network connection between nodes is down (such as a network partition, thinking back to the P in CAP). In that case, Mongo dictates that the majority of nodes that can still communicate now constitute the network.

MongoDB expects an odd number of total nodes in the replica set. Consider a five-node network, for example. If connection issues split it into a three-node fragment and a two-node fragment, the larger fragment has a clear majority and can elect a master and continue servicing requests. With no clear majority, a quorum couldn’t be reached.

To see why an odd number of nodes is preferred, consider what might happen to a four-node replica set. Say a network partition causes two of the servers to lose connectivity from the other two. One set will have the original master, but because it can’t see a clear majority of the network, the master steps down. The other set will similarly be unable to elect a master because it, too, can’t communicate with a clear majority of nodes. Both sets are now unable to process requests and the system is effectively down. Having an odd number of total nodes would have made this particular scenario—a fragmented network where each fragment has less than a clear majority—less likely to occur.

Some databases (such as CouchDB) are built to allow multiple masters, but Mongo is not, and so it isn’t prepared to resolve data updates between them. MongoDB deals with conflicts between multiple masters by simply not allowing them.

Because it’s a CP system, Mongo always knows the most recent value; the client needn’t decide. Mongo’s concern is strong consistency on writes, and preventing a multimaster scenario is not a bad method for achieving it.

Sharding

One of the core goals of Mongo is to provide safe and quick handling of very large datasets. The clearest method of achieving this is through horizontal sharding by value ranges—or just sharding for brevity. Rather than a single server hosting all values in a collection, some range of values is split, or sharded, onto other servers. For example, in our phone numbers collection, we may put all phone numbers less than 1-500-000-0000 onto Mongo server A and put numbers greater than or equal to 1-500-000-0001 onto server B. Mongo makes this easier by autosharding, managing this division for you.

Let’s launch a couple of (nonreplicating) mongod servers. Like replica sets, there’s a special parameter necessary to be considered a shard server (which just means this server is capable of sharding).

Now you need a server to actually keep track of your keys. Imagine you created a table to store city names alphabetically. You need some way to know that, for example, cities starting with A through N go to server mongo4 and O through Z go to server mongo5. In Mongo, you create a config server (which is just a regular mongod) that keeps track of which server (mongo4 or mongo5) owns what values. You’ll need to create and initialize a second replica set for the cluster’s configuration (let’s call it configSet).

Now enter the Mongo shell for the config server by running mongo localhost:27016 and initiate the config server cluster (with just one member for this example):

Finally, you need to run yet another server called mongos, which is the single point of entry for our clients. The mongos server will connect to the mongoconfig config server to keep track of the sharding information stored there. You point mongos to the replSet/server:port with the --configdb flag.

A neat thing about mongos is that it is a lightweight clone of a full mongod server. Nearly any command you can throw at a mongod you can throw at a mongos, which makes it the perfect go-between for clients to connect to multiple sharded servers. The following picture of our server setup may help.

Now let’s jump into the mongos server console in the admin database by running mongo localhost:27020/admin. We’re going to configure some sharding.

With that setup, now we have to give it the database and collection to shard and the field to shard by (in our case, the city name).

With all that setup out of the way, let’s load some data. If you download the book code, you’ll find a 12 MB data file named mongoCities100000.json that contains data for every city in the world with a population of more than 1,000 people. Download that file, and run the following import script that imports the data into your mongos server:

If the import is successful, you should see imported 99838 documents in the output (not quite 100,000 cities as the filename would suggest, but pretty close).

GeoSpatial Queries

Mongo has a neat trick built into it. Although we’ve focused on server setups today, no day would be complete without a little bit of razzle-dazzle, and that’s Mongo’s ability to quickly perform geospatial queries. First, connect to the mongos sharded server.

The core of the geospatial secret lies in indexing. It’s a special form of indexing geographic data called geohash that not only finds values of a specific value or range quickly but finds nearby values quickly in ad hoc queries. Conveniently, at the end of our previous section, we installed a lot of geographic data. So to query it, step 1 is to index the data on the location field. The 2d index must be set on any two value fields, in our case a hash (for example, { longitude:1.48453, latitude:42.57205 }), but it could easily have been an array (for example [1.48453, 42.57205]).

Now, we can use an aggregation pipeline (think back to Day 2) to assemble a list of all cities close to Portland, OR sorted in descending order by population (displaying the name of the city, the population, and the distance from the 45.52/-122.67 latitude/longitude point).

As you can see, Mongo’s aggregation API provides a very nice interface for working with schemaless geospatial data. We’ve only scratched the surface here (no pun intended), but if you’re interested in exploring the full potential of MongoDB, we strongly encourage you to dig more deeply.^[31]

GridFS

One downside of a distributed system can be the lack of a single coherent filesystem. Say you operate a website where users can upload images of themselves. If you run several web servers on several different nodes, you must manually replicate the uploaded image to each web server’s disk or create some alternative central system. Mongo handles this scenario using a distributed filesystem called GridFS.

Mongo comes bundled with a command-line tool for interacting with GridFS. The great thing is that you don’t have to set up anything special to use it. If you list the files in the mongos managed shards using the command mongofiles, you get an empty list.

But upload any file...

...and voilà! If you list the contents of mongofiles, you’ll find the uploaded name.

Back in your mongo console, you can see the collections Mongo stores the data in.

Because they’re just plain old collections, they can be replicated or queried like any other. Here we’ll look up the filename of the text file we imported.

Day 3 Wrap-Up

This wraps up our investigation of MongoDB. Today we focused on how Mongo enhances data durability with replica sets and supports horizontal scaling with sharding. We looked at good server configurations and how Mongo provides the mongos server to act as a relay for handling autosharding between multiple nodes. Finally, we toyed with some of Mongo’s built-in tools, such as geospatial queries and GridFS.

Day 3 Homework

Find

1. Read the full replica set configuration options in the online docs.
2. Find out how to create a spherical geo index.

Do

1. Mongo has support for bounding shapes (namely, squares and circles). Find all cities within a 50-mile radius around the center of London.^[32]

2. Run six servers: three servers in a replica set, and each replica set is one of two shards. Run a config server and mongos. Run GridFS across them (this is the final exam).