Weather of the Century: Part 4



Previously in this series

In part one of this series we introduced NOAA, the ISD data set, and the way we transformed the ISD data from a packed ASCII storage format into a MongoDB collection.

In part two we examined the Weather of the Century app, which uses the ISD to display weather information from all over the globe, from any point in time, on the Google Earth browser plugin. We discussed the queries the app uses in detail to get a sense for how MongoDB works with this data.

In part three, we introduced The Death Star and The Force, two very different kinds of MongoDB deployments, and compared their abilities to bulk load the ISD.

In this installment, we're going to conclude the comparison with a look at performance in two areas: everyday queries, and analytics and exploration. As in previous installments, we begin with a quick recap of our data model.

Data Recap

Here's what a weather observation record from the ISD looks like, stored in MongoDB. For a full discussion of this format, please refer to part one of this series.

  "st" : "u725053",
  "ts" : ISODate("2013-06-03T22:51:00Z"),
  "position" : {
    "type" : "Point",
    "coordinates" : [
  "elevation" : 231,
  "airTemperature" : {
      "value" : 21.1,
      "quality" : "1"
  "sky condition" : {
    "cavok": "N",
    "ceilingHeight": {
        "determination": "9",
        "quality": "1",
        "value": 1433
  "atmosphericPressure" : {
      "value" : 1009.7,
      "quality" : "5"

This data set contains 2.5 billion observations, with a total storage requirement of 4.5TB including indexes.

The Deployments

MongoDB Consulting Engineer André Spiegel assembled two deployments and put them both through their paces, using the ISD as a data set to work with. The first deployment is The Death Star, a single instance of the biggest storage-IO optimized VM that AWS offers. The second is The Force, a 100-shard cluster assembled from RAM-optimized instances of moderate size. For more detail, please review part three of this series.

Query Performance

In most cases, the queries an application issues to handle user requests must be able to use an index. Otherwise, the application would not perform well, nor would it support any kind of scale. Here we'll look at the performance of the two queries which are used in the Weather of the Century application, and another one in a similar vein.

Query 1: Point in Space-time

The first query is for a point in space-time, that is, a single observation from a single station. Let's say we want to know the weather at Kennedy Space Center at the time the Apollo 11 mission was launched. In our data set, that's a single document, retrievable directly through a single lookup in the ts_1_st_1 index.{"st" : "u747940",
              "ts" : ISODate("1969-07-16T12:00:00Z")})

For the performance trials, André issued a massive battery of these queries with randomized station ID and time.

For randomized single document retrieval, latency is very similar for The Death Star and The Force. Only 1/16th of the total data can fit into The Death Star's RAM at a time, so nearly all queries require paging, but with its mighty array of SSDs it can still answer in an average of 0.75ms. And while The Force has all the data in RAM ready to go, all queries must first be routed through mongos, adding a hop. These two factors cancel each other out on average, although we see The Force maintaining a tighter latency grouping into the 95th and 99th percentiles.

When we look at throughput, however, things are very different. The Death Star can maintain its response time up to a rate of 40,000 queries per second, but The Force, with its 100 mongods and 10 mongoses, achieves a staggering 610,000 q/s before its performance drop-off. (We define the achievable throughput as the number of queries that can be handled before the average response time becomes twice as long as for single queries.)

Query 2: One Station, One Year

The second query is for one single station, for a particular time period, like a single year. This query is not currently used in the Weather of the Century application, but it's a trivial feature set enhancement. The code in the example below is a query for the weather in Berlin for the entire year surrounding the fall of the Berlin wall. When it runs, the ts_1_st_1 index can be used to isolate the date range containing the correct documents, as well as the individual station.{"st" : "u103840",
              "ts" : {"$gte": ISODate("1989-01-01"),

                      "$lt" : ISODate("1990-01-01")}})

As with the first example, for these trials queries were issued with randomized station ID and year range.

To satisfy a query for every document for a single station for one year, MongoDB has to return a lot of documents. This is where we begin to see The Force, which has all its data in RAM, clobbers The Death Star, which might require hundreds of page faults to satisfy each query. The Death Star takes about 1.5 seconds on average, and for responses in the 95th and 99th percentile, between 3 and 4 seconds. Meanwhile, The Force averages 300ms, and even its worst responses, at just over a second, are faster than the fastest responses from The Death Star.

Throughput shows an even bigger difference. The Death Star can sustain 20 queries/s, which is respectable considering that any read parallelism it can achieve is hampered by a near guarantee that none of the needed documents are in memory. Meanwhile, The Force has all the data in RAM, and the query contains the station ID, which is The Force's shard key. So The Force can target the query to only the shard that contains relevant documents, and because that shard key is a hashed index, the distribution of queries to shards is extremely smooth. With all that working in its favor, The Force can handle 430 queries/s.

Query 3: The Whole World at One Moment

The third query is for a snapshot of the whole world at a particular moment in time, for example at the moment of the Y2K New Year. This query can also be satisfied by the ts_1_st_1 index, because the timestamp is a prefix of the compound index.{"ts" : ISODate("2000-01-01T00:00:00Z")})

For the performance trials, the queries where issued with randomized date-times.

This query contains no shard key, only the timestamp, so The Force must use a scatter/gather operation, querying every shard, and only assembling the final response once they have each completed their individual queries. Under many circumstances, scatter/gather operations introduce challenges to latency and throughput. In fact, the documentation concerning shard key selection makes clear that in general, the fastest queries target a single shard. During most scatter/gather queries, mongos waits for responses from shards that in the end have no documents to deliver, delaying responses. Additionally, those shards waste their time looking for documents that don't exist, reducing throughput. Does this mean The Death Star can take 1 out of 3 in André's query performance challenge?

As you can see, no, not by a long shot. While The Death Star may begin to stream its data after doing a single index lookup, it still triggers page faults for most of the blocks containing the response documents. For many years of the ISD, the number of stations in the world is greater than the number of samples a single station takes in a year, so the volume of documents to return is greater than that of the second query. As such, an average response time of 2 seconds is the best it can do, while the 95th percentile takes about 5 seconds, and the 99th is almost 8 seconds. This degree of "performance" can only be maintained up to 8 concurrent requests per second.

The Force, by contrast, is killing it, with an average latency of 200ms, a 95th percentile barely greater than that, and a 99th of about 1 second, half the time than the average response from The Death Star. It can keep this performance steady up to 310 queries/s.

The key to The Force's solid performance while issuing scatter/gather queries is the particular properties of the ISD as arranged by a hashed index of Station ID into our 100 shards. By virtue of this arrangement, it is reasonably certain that any query will require a significant quantity of work, and a roughly equivalent amount of it, on each shard. Rather than blocking every query to wait for 99 shards to come up with "no matches", The Force is delivering low-latency responses by efficiently parallelizing its work, and without losing much throughput to wasted time on shards.

Please note that this is a very special case, and if you plan to issue scatter/gather queries in production, you need to fully understand the mechanics involved. In general, please follow the advice found in the docs and ensure your applications queries can use the shard key to target their queries to a subset of shards.

Analytics and Exploration

Our third and final comparison is of how the two deployments handle queries that cannot use indexes. Every production database will face this type of load, some quite frequently. Without the ability to use indexes, MongoDB must scan every document in a collection to answer these queries.

There are two reasons why we might need to execute queries that can't use an index. Sometimes the particular kind of query was not anticipated when the system was built, and other times we can tolerate a high response time, so the overhead of an index would not be worth it. Consider that each ISD observation can record hundreds of values, all of which could be reasonable to query. Updating hundreds of indexes for each write operation would make performance awful, and storing hundreds of indexes would crush our data out of RAM. Indexes should only be added to speed up queries where minimized response times are required.

Find All Tornadoes

The query to find all tornadoes ever recorded is very simple. Our documents have a field called 'presentWeatherObservation.condition'. A value of '99' in that field means a tornado was present during the observation. This basic query finds any document matching that criterion. ({
  "presentWeatherObservation.condition" : "99"

The Death Star takes 1 hour, 28 minutes to answer that query. Without the benefit of an index, it must iterate over every single document in that 4.5TB collection, resulting in about 4 TB of paging. The Force, on the other hand, can execute this query in parallel on every shard, each of which has approximately 1/100th of the total data, and all of that already in RAM. It returns the answer in 47 seconds -- over 100x faster than The Death Star.

Maximum Temperature of All Recorded Measurements

To find the highest ever recorded temperature, we use the MongoDB aggregation pipeline framework. The $match phase of our pipeline filters out records not matching our QA standards, as described in part three of this series. In the $group phase, we treat the entire set as a single group by setting the _id value to null, and use the $max accumulator to select the greatest value. (That turns out to be 143° Fahrenheit, or 61.8° Celcius, which was measured in Death Valley.) ([
  { "$match" : { "airTemperature.quality" : 
                             { "$in" : [ "1", "5" ] } } },
  { "$group" : { "_id"     : null,
                 "maxTemp" : { "$max" :
                               "$airTemperature.value" } } }

It should come as no surprise that this type of query is the proton torpedo in the exhaust port of The Death Star. (You think we were going to come all this way using a Star Wars naming scheme and not eventually go there? Please. Be thankful I haven't named all 10 of those mongos instances after members of the Jedi Council.) With multiple value checks to perform, and some of its precious RAM devoted to the aggregation, The Death Star takes a whopping 4 hours, 45 minutes to answer this query. The Force, again, can race through its RAM in parallel on every shard, and finish with a trivial $max operation over 100 candidate numbers on the mongos. All this happens in 2 minutes -- almost 150x faster than on The Death Star. No targeting computer required.

In Summary

You can get away with hosting a lot of data on a single big server, when on a budget, if you only ever have to answer queries that can use an index. You can even get great latency and reasonable concurrency for those individual queries. However, you will not be able to build an index without impacting production for hours, nor should you expect any query requiring a collection scan to run in a reasonable time frame.

When the full flexibility of data exploration and large-scale service of concurrent users is required (of which at least one of is the case for many enterprises), horizontal scaling is called for. The cost may be ten times that of one large machine, but a hundredfold improvement of analytics performance may be the difference between a viable infrastructure and a useless one.

To be certain, most MongoDB deployments do not keep all data in RAM. Most enterprises will use a middle ground between the two extremes tested in this demonstration, one which meets both performance and budget requirements. As with all infrastructure planning, the key is experimentation.

If you’re interested in learning more about the performance best practices of MongoDB, download our guide:

Download Performance Best Practices

<< Read Part 3



About the Author - Avery

Avery is an infrastructure engineer, designer, and strategist with 20 years experience in every facet of internet technology and software development. As principal of Bringing Fire Consulting, he offers clients his expertise at the intersection of technology, business strategy, and product formulation. He earned a B.A in Computer Science from Brown University, where he specialized in systems and network programming, while also studying anthropology, fiction, cog sci, and semiotics. Avery got his start in internet technology in 1993, configuring apache and automating systems at Panix, the third-oldest ISP in the world. He has an obsession with getting to the heart of a problem, a flair for communication, and a devotion to providing delight to end users.