MongoDB のアグリゲーションパイプラインは、ドキュメントを $match（絞り込み）や $group（集計）などの「ステージ」に順に通して、集計結果や加工したデータを得る仕組みです。SQL の WHERE、GROUP BY、JOIN に当たる処理を、配列で並べたステージの組み合わせで書きます。この記事では、注文データを例に、よく使うステージの書き方、配列の展開と別コレクションとの結合、インデックスを効かせる並べ方と実行計画の確認、結果の保存までを説明します。
パイプラインの考え方
aggregate() には、ステージの配列を渡します。各ステージは前のステージから受け取ったドキュメントを加工して、次のステージに渡します。
db.orders.aggregate([
{ $match: { status: "paid" } }, // 1. 絞り込む
{ $group: { _id: "$customerId", total: { $sum: "$total" } } }, // 2. 集計する
{ $sort: { total: -1 } }, // 3. 並べ替える
{ $limit: 10 } // 4. 件数を絞る
])"$customerId" のように先頭に $ が付いた文字列は「そのフィールドの値」を表します。ステージ名の $match と、値の参照の $ は意味が異なるので、最初はここで混乱しやすい点です。
この記事では、orders コレクションに次のようなドキュメントが入っている前提で説明します。
{
_id: ObjectId("..."),
customerId: ObjectId("..."),
status: "paid",
total: 5800,
items: [
{ sku: "TSHIRT-M", qty: 2, price: 1900 },
{ sku: "CAP-01", qty: 1, price: 2000 }
],
createdAt: ISODate("2026-10-03T02:15:00Z")
}コマンドは mongosh（MongoDB Shell）で実行する形で書いています。演算子によって使えるようになったバージョンが異なり、この記事では MongoDB 5.0 以降で使えるものを使っています。現在サポートされているバージョンは公式サイトで確認してください（2026年10月現在、8.x 系が中心です）。
よく使うステージ
最初に覚えておくと多くの集計を書けるのは、次のステージです。
$match：条件に合うドキュメントだけを通す。find()と同じ条件の書き方が使える$group：_idに指定したキーごとにまとめ、$sum、$avg、$max、$min、$push、$addToSetなどで集計する$project：出力するフィールドを選ぶ、または計算したフィールドを作る$set（$addFieldsと同じ）：既存のフィールドを残したまま、フィールドを追加・上書きする$sort、$limit、$skip：並べ替えと件数の調整$count：通ってきたドキュメントの件数を数える
月ごとの売上を集計する
日付を月単位に切り捨ててからまとめます。$dateTrunc は 5.0 から使える演算子です。
db.orders.aggregate([
{ $match: {
status: "paid",
createdAt: {
$gte: new Date("2026-01-01T00:00:00+09:00"),
$lt: new Date("2027-01-01T00:00:00+09:00")
}
} },
{ $group: {
_id: { $dateTrunc: { date: "$createdAt", unit: "month", timezone: "Asia/Tokyo" } },
sales: { $sum: "$total" },
orders: { $sum: 1 },
avgOrder: { $avg: "$total" }
} },
{ $sort: { _id: 1 } }
])MongoDB の日付は UTC で保存されます。日本時間の「月」で区切りたい場合は、$match の範囲を日本時間の時刻で指定し、$dateTrunc にも timezone を指定します。どちらかが抜けると、月初の9時間分のデータが前の月に入るといったずれが起きます。{ $sum: 1 } は、1件ごとに1を足す、つまり件数を数える書き方です。
配列を展開して集計する（$unwind）
1つの注文に複数の商品が入っている場合、商品ごとの販売数を出すには、配列 items を1要素1ドキュメントに展開します。
db.orders.aggregate([
{ $match: { status: "paid" } },
{ $unwind: "$items" },
{ $group: {
_id: "$items.sku",
qty: { $sum: "$items.qty" },
sales: { $sum: { $multiply: ["$items.qty", "$items.price"] } }
} },
{ $sort: { qty: -1 } },
{ $limit: 10 }
])$unwind の後は、items が配列ではなく1つの商品のオブジェクトになります。配列が空の注文や items がない注文は、既定では結果から消えます。残したい場合は { $unwind: { path: "$items", preserveNullAndEmptyArrays: true } } と書きます。
展開すると件数が「注文数 × 商品数」に増えるので、$unwind の前に $match で対象を絞っておくことが、処理を軽くするうえで大切です。
別のコレクションと結合する（$lookup）
注文に顧客の名前を付けるには、$lookup で customers コレクションと結合します。
db.orders.aggregate([
{ $match: { status: "paid", total: { $gte: 10000 } } },
{ $lookup: {
from: "customers",
localField: "customerId",
foreignField: "_id",
as: "customer"
} },
{ $unwind: "$customer" },
{ $project: { _id: 0, total: 1, createdAt: 1, "customer.name": 1 } }
])$lookup の結果は、条件に合ったドキュメントの配列として as に指定したフィールドに入ります。1件だけ対応する場合でも配列になるので、続けて $unwind で展開すると扱いやすくなります。
結合先の foreignField にインデックスがないと、注文1件ごとに顧客のコレクション全体を調べることになり、件数が増えると大きく遅くなります。また、結合した結果を含めた1つのドキュメントは 16MB の上限を超えられません。1件に大量のドキュメントが結合される場合は、$lookup の中にパイプラインを書いて必要なフィールドや件数に絞るか、データの持ち方を見直します。
MongoDB では、頻繁に一緒に読むデータは1つのドキュメントに埋め込むことが推奨される場面も多く、毎回 $lookup が必要になる設計は見直しの合図になることがあります。
インデックスを効かせる並べ方と実行計画
パイプラインの先頭にある $match と $sort は、コレクションのインデックスを使えます。逆に、$group や $project の後に置いた $match は、加工後のデータに対する絞り込みになるため、インデックスは使われません。MongoDB はステージの順序を自動で入れ替えて最適化することもありますが、基本は「先に絞る、次に加工する」の順に書きます。
月ごとの売上の例であれば、次の複合インデックスが $match に効きます。
db.orders.createIndex({ status: 1, createdAt: 1 })インデックスが使われているかは、explain で確認します。
db.orders.explain("executionStats").aggregate([
{ $match: { status: "paid", createdAt: { $gte: new Date("2026-10-01T00:00:00+09:00") } } },
{ $group: { _id: null, sales: { $sum: "$total" } } }
])出力の中に IXSCAN（インデックスを使った検索）があれば、インデックスが使われています。COLLSCAN になっている場合は、コレクション全体を読んでいます。totalDocsExamined（調べたドキュメント数）が、結果の件数に比べて極端に多くないかも見ておきます。本番のデータで試す前に、データ量の近いステージング環境で確認すると安全です。
メモリにも制限があります。$group や $sort などのステージは、それぞれ 100MB を超えるとディスクの一時ファイルを使う必要があります。MongoDB 6.0 以降では、既定でディスクを使って続行する設定になっていますが、それより前のバージョンでは aggregate() に { allowDiskUse: true } を指定しないとエラーになります。ディスクを使うと遅くなるため、まず先頭の $match で件数を減らすことを考えます。
結果を保存する（$merge と $out）
毎回重い集計をするのではなく、日次や月次の集計結果をコレクションに保存しておき、画面ではそれを読むようにすると負荷を下げられます。
db.orders.aggregate([
{ $match: { status: "paid", createdAt: { $gte: new Date("2026-10-01T00:00:00+09:00") } } },
{ $group: {
_id: { $dateTrunc: { date: "$createdAt", unit: "day", timezone: "Asia/Tokyo" } },
sales: { $sum: "$total" }
} },
{ $merge: { into: "daily_sales", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } }
])$merge はパイプラインの最後に置き、指定したコレクションに結果を追加・更新します。同じ日の集計をやり直しても、_id が一致する行が置き換わるだけなので、定期的に実行する処理に向いています。$out は出力先のコレクションを結果で丸ごと置き換えるため、既存のデータを残したい場合は $merge を使います。どちらも書き込みを伴うので、出力先の名前を間違えないように注意してください。
アプリケーションから使うときの注意
Python の PyMongo や Node.js のドライバーでも、aggregate() にステージの配列を渡す書き方は同じです。結果はカーソルとして返るので、件数が多い場合は一度にリストへ変換せず、順に読み込みます。
検索フォームの入力などをパイプラインに入れる場合は、受け取った値を文字列や数値に変換してから使ってください。リクエストの JSON をそのまま $match に入れると、{ "$ne": null } のような演算子を送り込まれ、意図しない条件で検索されるおそれがあります（オペレーターインジェクション）。
まとめ
- アグリゲーションパイプラインは、ステージの配列にドキュメントを順に通して集計・加工する仕組み
$match→$group→$sortが基本の形。日本時間で集計するときはtimezoneの指定を忘れない- 配列は
$unwind、別コレクションとの結合は$lookup。結合先のフィールドにはインデックスを作る $matchと$sortを先頭に置き、explain("executionStats")でIXSCANになっているかを確認する- 重い集計は
$mergeで結果を保存し、利用者の入力は型を変換してからパイプラインに入れる