Intro

At Up to Eleven we are currently using MongoDB 2.0 in a sharded setup. Where each shard has 3 replica sets.

The biggest collection stores messages to a specific address with a read flag, status and a message date.

For the web we need an overview of this messages per address (conversation). Therefore we currently use a mapreduce command,  because in a sharded setup the group function is not available.

With release of MongoDB 2.2 their are now 2 new options for this problem – It’s now possible to run MapReduce commands with JsMode on shards and their also is the brand new aggregation framework.

In the following article the options are compared to each other.

Schema

Document

{
  "_id" : NumberLong("1073741825094"), // combination of message id & user id
  "u" : 250, // user id
  "a" : "test", // address
  "b" : "Hello, what are you doing right now?", // body
  "i" : true, // incoming flag
  "r" : true, // read flag
  "s" : 0, // status
  "d" : ISODate("2011-06-06T18:49:55Z") // date
}

{
  "v" : 1,
  "key" : {
    "_id" : 1
  },
  "ns" : "test.user_messages",
  "name" : "_id_"
},
{
  "v" : 1,
  "key" : {
    "u" : 1,
    "a" : 1,
    "d" : -1
  },
  "ns" : "test.user_messages",
  "name" : "conversation"
}

Tests

MapReduce

db.runCommand({
  "mapreduce": "user_messages",
  "map": "function(){
     emit(this.a, {
       count:1,
       unread:this.r ? 0 : 1,
       unsent:this.s==5 ? 1 : 0,
       messageId:this._id,
       date:this.d,
       body:this.b
    });
  }",
  "reduce": "function(address,values) {
    var result = { 
      count:0,
      unread:0,
      unsent:0,
      messageId:0,
      date:0,
      body:''
    };
    values.forEach(function(value) {
      result.count += 1;
      if (value.unread > 0) result.unread += 1;
      if (value.unsent > 0) result.unsent += 1;
      if (value.messageId > result.messageId) result.messageId = value.messageId;
      if (value.dateSent >= result.dateSent) {
        result.message = value.message;
        result.dateSent = value.dateSent;
      }
    });
    return result;
  }",
  "verbose" : true,
  "out" : { "inline" : 1 },
  "query" : { "u" : 1 },
}

time: 38 s 718 ms

MapReduce with JsMode

db.runCommand({
  "mapreduce": "user_messages",
  "map": "function(){
     emit(this.a, {
       count:1,
       unread:this.r ? 0 : 1,
       unsent:this.s==5 ? 1 : 0,
       messageId:this._id,
       date:this.d,
       body:this.b
    });
  }",
  "reduce": "function(address,values) {
    var result = { 
      count:0,
      unread:0,
      unsent:0,
      messageId:0,
      date:0,
      body:''
    };
    values.forEach(function(value) {
      result.count += 1;
      if (value.unread > 0) result.unread += 1;
      if (value.unsent > 0) result.unsent += 1;
      if (value.messageId > result.messageId) result.messageId = value.messageId;
      if (value.dateSent >= result.dateSent) {
        result.message = value.message;
        result.dateSent = value.dateSent;
      }
    });
    return result;
  }",
  "verbose" : true,
  "out" : { "inline" : 1 },
  "query" : { "u" : 1 },
  "jsMode" : true
}

time: 22 s 237 ms

Aggregation Framework

db.user_messages.aggregate(
 { $match: { u:1 } },
 { $sort: { a:1, d:-1 } },
 { $group: {
     _id: "$a",
     count: { $sum : 1 },
     unread: { $sum : { $cond : [ "$r", 0 , 1]} },
     unsent: { $sum : { $cond : [ { $eq : [ "$s", 5 ] }, 1, 0 ] } },
     messageId: { $max : "$_id" }, 
     date: { $max : "$d" },
     body: { $first : "$b" },
 } }
);

time: 6 s 835 ms

Conclusion