MongoDBのAggregationを使ってみる

MongoDBでAggregation=集計を行う方法は3つあります。

1. Aggregation Pipeline
2. map-reduce function
3. single purpose aggregation method

今回はこのうちのAggregation Pipeline(以下、pipeline)を使ってみました。pipelineがMongoDBで集計を行う上では好ましい方法であるとされています。

pipelineの使い方

実例を挙げてみます。orderコレクションの中に次のようなデータがあったとします。_id列は省略してあります。

> db.order.find()
{ "cust_id" : "A123", "amount" : 500, "status" : "A" }
{ "cust_id" : "A123", "amount" : 250, "status" : "A" }
{ "cust_id" : "B212", "amount" : 200, "status" : "A" }
{ "cust_id" : "A123", "amount" : 300, "status" : "D" }

statusがAであるレコードを取り出して、さらにcust_idが同じもの同士でamountの合計を表示するという集計をしたいとき、pipelineでは次のように書くことができます。

db.order.aggregate([
    { $match: { "status": "A" } },
    { $group: { _id: "$cust_id", total: { $sum: "$amount" } } }
])

$matchでフィルターをかけ、$groupで表示結果のキーの設定と集計方法の設定を行っているのがわかると思います。pipelineはその名の通り長いパイプで、MongoDB内のデータがこのパイプを通って出てくると集計結果になります。パイプの途中にはフィルターや他のパイプとの合流地点があり、このようにパイプの中でデータの操作を行う部分をpipelineの"stage"といいます。上の例では{ $match: { "status": "A" } }や{ $group: { _id: "$cust_id", total: { $sum: "$amount" } } }がstageに当たり、$matchや$groupをさらに"stage operator"と呼びます。
pipelineはdb.collections.aggregate()に配列でstageを渡していくだけで使えるのですが、stageを渡す順番に注意してください。pipelineは上から順番に処理されます。上の例で2つのstageを逆にすると集計がされなくなります。

簡単な集計

次のテストデータで集計を試してみます。人口は適当です。

> db.cities.insertMany([
    {"name": "sapporo",   "area": "hokkaido", "population": 3},
    {"name": "sendai",    "area": "tohoku",   "population": 5},
    {"name": "yamagata",  "area": "tohoku",   "population": 1},
    {"name": "tokyo",     "area": "kanto",    "population": 10},
    {"name": "yokohama",  "area": "kanto",    "population": 7},
    {"name": "chiba",     "area": "kanto",    "population": 4},
    {"name": "nagoya",    "area": "tokai",    "population": 6},
    {"name": "shizuoka",  "area": "tokai",    "population": 3},
    {"name": "kobe",      "area": "kansai",   "population": 5},
    {"name": "osaka",     "area": "kansai",   "population": 8},
    {"name": "hiroshima", "area": "chugoku",  "population": 5},
    {"name": "okayama",   "area": "chugoku",  "population": 4},
    {"name": "matsuyama", "area": "shikoku",  "population": 2},
    {"name": "fukuoka",   "area": "kyushu",   "population": 7},
    {"name": "kagoshima", "area": "kyushu",   "population": 2},
    {"name": "miyazaki",  "area": "kyushu",   "population": 1},
    {"name": "naha",      "area": "okinawa",  "population": 1}
])

pipelineは{}を書くことが多いので、コマンドライン上だと{}の対応がわからなくなります。jsファイルを作ってmongoコマンドにDB名と一緒に渡してあげるとわかりやすくなります。

function get_results (result) {
    print(tojson(result));
}

db.cities.aggregate([
    // ここにstageを書く
]).forEach(get_results)

実行するときは

$ mongo testDB mongo.js

$match、$project、$sort

$matchでフィルターがかけられます。数値の比較にも専用の演算子があるので注意してください。$projectは集計結果の表示を指定できます。$sortは並び順です。1を指定すると昇順、-1を指定すると降順に並べ替えてくれます。

function get_results (result) {
    print(tojson(result));
}

print('関東地方')
db.cities.aggregate([
    { $match: { "area": "kanto" } },
]).forEach(get_results)

print('関東地方、_id列以外')
db.cities.aggregate([
    { $match: { "area": "kanto" } },
    { $project: { _id: 0 } },
]).forEach(get_results)

print('人口が5以上、area列とpopulation列だけ')
db.cities.aggregate([
    { $match: { "population": { $gte: 5 } } },
    { $project: { _id: 0, area: 1, population: 1} },
    { $sort: { name: 1,  } }
]).forEach(get_results)

print('人口が5以上、_id列以外、populationの多い順')
db.cities.aggregate([
    { $match: { "population": { $gte: 5 } } },
    { $project: { _id: 0, name: 1, area: 1, population: 1} },
    { $sort: { population: -1 } },
]).forEach(get_results)

print('人口が5以上かつ8未満、_id列以外、populationの少ない順')
db.cities.aggregate([
    { $match: { "population": { $gte: 5 , $lt: 8} } },
    { $project: { _id: 0, name: 1, area: 1, population: 1} },
    { $sort: { population: 1 } },
]).forEach(get_results)

$sample、$limit

$sampleはランダムにレコードを取得することができます。$limitは次のステージに渡すレコード数を制限できます。

function get_results (result) {
    print(tojson(result));
}

print('ランダムに5レコード、_id列以外、name順')
db.cities.aggregate([
    { $sample: { size: 5 } },
    { $project: { _id: 0 } },
    { $sort: { name: 1} }
]).forEach(get_results)

print('3レコードに制限、ランダムに5レコード、_id列以外、name順')
db.cities.aggregate([
    { $limit: 3 },
    { $sample: { size: 5 } },
    { $project: { _id: 0 } },
    { $sort: { name: 1} }
]).forEach(get_results)

$count

$countはstageに残っているレコード数を指定したキーで表示します。

function get_results (result) {
    print(tojson(result));
}

print('人口が7より多い')
db.cities.aggregate([
    { $match: { "population": { $gt: 7 } } },
    { $count: "big city" }
]).forEach(get_results)

簡単なところは以上です。stageの順番によって結果が変わってくるというのが一番の注意点かなと思います。