MongoDBのMapReduceを使ってみたときのメモ

■MapReduce の決まり
・map()でemit()した値の形式とreduce()の戻り値の形式が一致していること reduce関数は一度のMapReduce内で何度も呼ばれ、全てのreduceの戻り値をreduceした結果がMapReduceの戻り値となる。
→最終的な結果 = reduce(key, [values_3, reduce(key, [values_2, reduce(key, [values_1])])])

また、map()内で1件しかemit()されなかったkeyと値は、reduce()で一度も処理されない!!

すべてreduceした後の値を処理したい場合は、finalize functionを利用できる。

mongoの公式マニュアルが詳しい。
MapReduce - Docs-Japanese - 10gen Confluence

(mongoは集計できるので)mongoでは使いどころは限られると思うが、簡単にjavascriptでMapReduceできるなんてMongoDBやっぱいいね(･∀･)


■ MapReduceでやる意味(長所)
・集計対象のデータが巨大になっても負荷分散できる。
・oracleのマテビューのように結果を保存できる。(merge, replaceも可能)
・そもそもcassandraやhbaseでは集計関数やキー以外の条件検索ができないが、MapReduceなら集計できる。
そのため集計用のテーブルなどを別に作成せず、ユーザーのトランザクションデータを直接集計できる。


■ データの初期化

// データの初期化
db.userActivity.remove()

// 登録
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/06 1:00:00')});
db.userActivity.insert({name: 'zoro', type: 'login', at: new Date('2012/05/06 2:00:00')});
db.userActivity.insert({name: 'usopp', type: 'login', at: new Date('2012/05/06 3:00:00')});
db.userActivity.insert({name: 'nami', type: 'login', at: new Date('2012/05/06 4:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/05 5:00:00')});
db.userActivity.insert({name: 'zoro', type: 'login', at: new Date('2012/05/05 6:00:00')});
db.userActivity.insert({name: 'nami', type: 'login', at: new Date('2012/05/05 7:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/04 8:00:00')});
db.userActivity.insert({name: 'usopp', type: 'login', at: new Date('2012/05/04 9:00:00')});
db.userActivity.insert({name: 'nami', type: 'login', at: new Date('2012/05/04 10:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/03 11:00:00')});
db.userActivity.insert({name: 'zoro', type: 'login', at: new Date('2012/05/03 12:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/02 13:00:00')});
db.userActivity.insert({name: 'zoro', type: 'login', at: new Date('2012/05/02 14:00:00')});
db.userActivity.insert({name: 'usopp', type: 'login', at: new Date('2012/05/02 15:00:00')});
db.userActivity.insert({name: 'nami', type: 'login', at: new Date('2012/05/02 16:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'login', at: new Date('2012/05/01 17:00:00')});
db.userActivity.insert({name: 'usopp', type: 'login', at: new Date('2012/05/01 18:00:00')});

db.userActivity.insert({name: 'ruffy', type: 'event_login', at: new Date('2012/05/06 19:00:00')});
db.userActivity.insert({name: 'nami', type: 'event_login', at: new Date('2012/05/06 20:00:00')});
db.userActivity.insert({name: 'zoro', type: 'event_login', at: new Date('2012/05/05 21:00:00')});
db.userActivity.insert({name: 'nami', type: 'event_login', at: new Date('2012/05/05 22:00:00')});
db.userActivity.insert({name: 'usopp', type: 'event_login', at: new Date('2012/05/04 23:00:00')});
db.userActivity.insert({name: 'nami', type: 'event_login', at: new Date('2012/05/04 1:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'event_login', at: new Date('2012/05/03 2:00:00')});
db.userActivity.insert({name: 'zoro', type: 'event_login', at: new Date('2012/05/03 3:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'event_login', at: new Date('2012/05/02 4:00:00')});
db.userActivity.insert({name: 'nami', type: 'event_login', at: new Date('2012/05/02 5:00:00')});
db.userActivity.insert({name: 'ruffy', type: 'event_login', at: new Date('2012/05/01 6:00:00')});


// 検索
db.userActivity.find({name: 'ruffy'}).sort({at:-1});
db.userActivity.find({name: 'zoro'}).sort({at:-1});

■ 誤った方法
MapReduceの決まりに沿っていないため、動きはするがデータ量が増えるとNG
決まり：map()でemit()した値の形式とreduce()の戻り値の形式が一致していること

// ------------------------------------------------
// ユーザー軸の集計
map = function () {
  // 'ruffy', {type: login, dt: 2012-05-05}
  emit(this.name, {'type': this.type, 'dt': this.at});
}
reduce = function(key, values) {
  var result = {login: [], event_login: []};
  values.forEach(function(value) {
    type = value.type
    dt = value.dt
    // javascriptの日付操作、なんとかならんかの。
    dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
    result[type].push(dt)
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});

// とはいえ、これはユーザーでorderすれば同じようなことができる
db.userActivity.find().sort({name: 1, at: -1, type:1})


// ------------------------------------------------
// 日付軸の集計
map = function () {
  dt = this.at;
  dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
  // 2012-05-05, {type: login, name: 'ruffy'}
  emit(dt, {'type': this.type, 'name': this.name});
}
reduce = function(key, values) {
  var result = {login: [], event_login: []};
  values.forEach(function(value) {
    type = value.type
    result[type].push(value.name)
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});

// sqlだとdate型のカラムがない状態で集計は面倒だけど楽にできちゃうヽ(・∀・)ﾉ

// 対象データをfilterして集計とかもできちゃう
db.userActivity.mapReduce(map, reduce, {query: {at: {$gte: new Date('2012/05/03')}}, out: {inline : 1}});


// ------------------------------------------------
// 活動軸の集計
map = function () {
  // login, {name: 'ruffy', dt: 2012-05-05}
  emit(this.type, {'name': this.name, 'dt': this.at});
}
reduce = function(key, values) {
  var result = {};
  values.forEach(function(value) {
    dt = value.dt
    dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
    if (typeof result[dt] == 'undefined') {
      result[dt] = 0
    }
    result[dt] += 1
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});

db.userActivity.mapReduce(map, reduce, {query: {type: 'login'}, out: { inline : 1}});

■ 正しい方法

// ------------------------------------------------
// ユーザー軸の集計
map = function () {
  // 'ruffy', {type: login, dt: 2012-05-05}
  dt = this.at
  dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
  value = {}
  value[this.type] = [dt]
  emit(this.name, value);
}
reduce = function(key, values) {
  var result = {};
  values.forEach(function(value) {
    for (type in value) {
      for (i in value[type]) {
        if (typeof result[type] == 'undefined') { result[type] = []; }
        dt = value[type][i]
        result[type].push(dt)
      }
    }
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});


// ------------------------------------------------
// 日付軸の集計
map = function () {
  dt = this.at
  dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
  value = {}
  value[this.type] = [this.name]
  emit(dt, value);
}
reduce = function(key, values) {
  var result = {};
  values.forEach(function(value) {
    for (type in value) {
      for (i in value[type]) {
        if (typeof result[type] == 'undefined') { result[type] = []; }
        dt = value[type][i]
        result[type].push(dt)
      }
    }
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});



// ------------------------------------------------
// 活動軸の集計
map = function () {
  dt = this.at
  dt.setHours(0);dt.setMinutes(0);dt.setSeconds(0);dt.setMilliseconds(0);
  value = {}
  value[dt] = 1
  emit(this.type, value);
}
reduce = function(key, values) {
  var result = {};
  values.forEach(function(value) {
    for (dt in value) {
      if (typeof result[dt] == 'undefined') { result[dt] = 0; }
      count = value[dt];
      result[dt] += parseInt(count)
    }
  });
  return result;
}
// 実行(inlineオプションは、結果を保存せず画面出力するオプション)
db.userActivity.mapReduce(map, reduce, {out: { inline : 1}});

■ 参考サイト MongoDBのMapReduceは、その他にも様々なオプションでいろいろできる!
MapReduce - Docs-Japanese - 10gen Confluence - MongoDB