MongoDB の Aggregation Framework でラクラク集計生活 (2.6 対応) #MongoDB

どんなお話？

MongoDB 2.6 で使い放題になった Aggregation Framework で集計してみたよ
簡単！(笑)
MongoDBのデータを集計するなら、 Aggregation Framework がファーストチョイス

前置き

自分が関わった MongoDB を使ったサービス(某ウザいと評判のサービス)が最近リリースされました。
MongoDBに不定形データを保存するのですが、開発は昨年からスタートされており、
途中で MongoDB が 2.6 にメジャーバージョンアップされていろいろ変わった中で、
今回は集計にフォーカスして Aggregation Framework を使ったお話を。
既に MongoDB をバリバリ利用している方にとっては今更感は否めませんが、未使用の方に雰囲気を感じていただければ幸い。

2.6 以前

集計といえば MapReduce よねという感じで MapReduce を使っていたのですが、複雑な集計が可能な反面、

遅い
Map 用関数、Reduce 用関数の管理が面倒(パラメータをちょっとずつ変えて、似たような集計を何回も集計したいときに頭にくる)

といった不満がありました。
しかし 2.6 以前にあった Aggregation Framework (2.2 で登場)は

記述が( MapReduce に比べて)簡潔で集計用パラメータの流用が容易

という利点はありましたが

16MB 制限がしんどい(確実に合計 16MB を超える数十万レコードから集計したかった)

という致命的な弱点から採用ができず、やっぱり MapReduce だよなーということになっていました。

2.6 登場

そして時が過ぎ 2014 年 4 月。 MongoDB 2.6 がリリースされ、しばらくしてから(バージョンアップするかというタイミングになって)気がつくわけです。
「 Aggregation Framework 、容量制限なくなってね？」と。(本来ならきちんとキャッチアップして検証しておくべきなんですが)
普通に集計する分には MapReduce より有意に速い。パイプラインを工夫すれば MapReduce した時に似せた出力結果にも加工できる。
処理もパイプラインで簡単なフィルタオブジェクトを直列に連結するだけ。見通しがよく管理もしやすい。
ということで Aggregation Framework を使い出してみたという話です。今まで無視しててごめんね。

まー、 16MB 制限下で十分使えていた人にしてみたら「何を今更」な感じかもしれません。

こんな感じで使ってみた

その 1

{
  service: "hoge",
  answer: 2,
  user: {
    age: 18,
    gender: 1
  }
  startTIme: ISODate(～～),
  endTime: IDODate(～～)
}

こんな感じのデータがlogコレクションにいっぱい入ってると想定して、
hoge サービスのアクセス時間( endTime - StartTime )を5秒単位で男女( user.gender )別に集計したい時はこのように aggregate します。

db.log.aggregate([
  {$project: {
    service: "$service",
    gender: "$user.gender",
    time: {
    $divide: [
      {$subtract: [
        {$subtract: ["$endTime", "$startTime"]},
        {$mod: [{$subtract: ["$endTime", "$startTime"]}, 5]}
      ]},
      5]}
  }},
  {$match: {
    "service": "hoge"
  }},
  {$group: {
    _id: {
      time: {$multiply: ["$time", 5]},
      gender: "$gender"
    },
    count: {$sum: 1}
  }},
  {$sort: {
    "_id.gender": -1
  }},
  {$group: {
    _id: {time: "$_id.time"},
    row: {"$addToSet": {
      gender: "$_id.gender",
      count: "$count"
    }},
    countSum: {"$sum": "$count"}
  }},
  {$sort: {
    "_id": 1
  }}
]);

結果は

{ "_id" : { "time" : 0 }, "row" : [ { "gender" : 1, "count" : 100 }, { "gender" : 2, "count" : 50 } ], "countSum" : 150 }
{ "_id" : { "time" : 5 }, "row" : [ { "gender" : 1, "count" : 72 }, { "gender" : 2, "count" : 36 } ], "countSum" : 108 }
{ "_id" : { "time" : 10 }, "row" : [ { "gender" : 1, "count" : 60 }, { "gender" : 2, "count" : 26 } ], "countSum" : 86 }
{ "_id" : { "time" : 15 }, "row" : [ { "gender" : 1, "count" : 40 }, { "gender" : 2, "count" : 12 } ], "countSum" : 52 }
...

こんな感じ。 5 秒毎の gender = 1, gender = 2 のそれぞれの集計結果と、男女合計が取れます。
集計対象が 10 万件だろうが 100 万件だろうが、(データ量に応じて相応の時間をかければ)へっちゃらです。

何やってるの？

aggregate パイプラインの流れの基本は

$project ステージで集計に使うデータ形式を定義
$match ステージで集計対象の抽出
\$group ステージで集計(集計後データの定義)
$sort ステージで並び替え
( \$out ステージでコレクションに書き出し( \$out も 2.6 で利用可能になりました))

って感じになります。各ステージは任意の順番で実行でき、回数にも制限はありません
(扱うデータが大きいままのステージの数が多くなれば、それ相応に実行時間がかかります)。
上の例では $group ステージを2回行って、1回目に男女別の集計、2回目に男女の合計の算出をしています。

で、 \$divide, \$subtract, \$mod が交錯しているところは、5秒ごとにまとめてるところです。
普通に \$subtract すると、1/5 で 0.2 になってしまうので、余りを減算して全部整数になるようにしています。よくある罠回避ですね。

その 2

{
  "id": 1,
  "name": "moge",
  "gender": 1,
  "using": [1, 4, 5, 6, 9]
}

こんな感じのデータが入っている user コレクションに対して、using に入っている値を集計します。

db.user.aggregate([
  {$project: {
    "using": "$using",
  }},
  {$group: {
    _id: {},
    using: {$push: "$using"},
    count: {"$sum": 1}
  }}
  {$unwind: "$using"},
  {$unwind: "$using"},
  {$group: {
    _id: "$using",
    countAll: {$first: "$count"},
    count: {"$sum": 1}
  }}
]);

という処理を実行。今回は全件集計するので \$match ステージがありません。1回目の \$group ステージで using をひとまとめにしつつ
(配列を push しているので二重配列になります)
user の件数を集計します。その後の2回の \$unwind ステージで
using の二重配列を展開し、
2回目の \$group ステージで using のそれぞれの値を集計します。

{ "_id" : 1, "countAll": 360, "count": 170 }
{ "_id" : 2, "countAll": 360, "count": 140 }
{ "_id" : 3, "countAll": 360, "count": 12 }
{ "_id" : 4, "countAll": 360, "count": 310 }
...

結果こうなります。 countAll を用意するのは、何人中何人 using にその値が入っているのかを集計結果でわかるようにするためです。 countAll はひとつでいいじゃんとなったら、
その 1 同様、もう一段 $group ステージを用意すれば良いでしょう。パーセンテージ欲しいなという場合は、 \$project ステージで割合を計算( count / countAll )することができます。

その 3

最後にもう一つ例を出してみましょう。
MongoDB でありがちな、「特定のデータがあったりなかったりするコレクション」の、そのデータを集計したい場合。

{
  id: 1,
  age: 18,
  gender: 1
  wear: {
    cap: 1
  }
},
{
  id: 2,
  age: 18,
  gender: 1
  wear: {
    glasses: 3
  }
},
...

まあこんな user コレクションがあるとして、 wear 下の構造が人それぞれな場合に、メガネをかけている( wear.glasses に何らかの値が入っている)人を数えたい。そんな場合。

db.response.aggregate([
  {$match: {
    "wear.glasses": {$exists: true}
  }},
  {$group: {
    _id: {},
    count: {"$sum": 1}
  }}
]);

これでOK。キモは \$match ステージでの $exists ですね。 wear.glasses が存在するレコードだけ集計対象にします。
メガネの種類( wear.glasses の値)別に集計したいときは

db.response.aggregate([
  {$match: {
    "wear.glasses": {$exists: true}
  }},
  {$group: {
    _id: "$wear.glasses",
    count: {"$sum": 1}
  }}
]);

\$group ステージで _id を指定すれば、それの値別に集計できます。簡単ですね。
この場合、 \$match ステージの条件を外して集計すると、 wear.glasses が定義されていない分は id が null として集計されますので、ご利用シーンに合わせて使い分けしましょう
( null が出てくるのが嫌な場合は前段に \$project ステージを入れ、そこで wear.glasses がなかったら -1 (とか任意の値)を入れておくなどすると良いです(あ、 \$cond や \$ifNull について説明してないですね。詳細はリファレンスを見ていただくとして、 \$cond は if 文というか、?: 三項演算子的なもので、これを使えばいろいろ仕込めます。 \$ifNull は見たままです。))

まとめ

MongoDB で集計するなら Aggregation Framework の利用をまず検討しましょう。 MapReduce? 一旦忘れましょう。くれぐれも 2.6 にバージョンアップしていないなんてことのないように。最新の安定バージョンは 2.6.6 (2014/12/9 リリース)ですよ。

MongoDB の Aggregation Framework でラクラク集計生活 (2.6 対応)