【MongoDB】集約パイプライン（Aggregation Pipeline）の基本と実践的な使い方

MongoDB
2023.12.16

こんにちは、かつコーチです。

MongoDBでの集計処理といえば、まず名前が挙がるのが集約パイプライン（複数の処理ステージを連結してドキュメントを段階的に加工・集計する仕組み）です。

SQLのGROUP BYに慣れている方ほど、最初にこの仕組みの発想の違いに戸惑うはずです。

この記事は前提知識ありきの上級者向け記事として、パイプラインの構造から実務でつまずきやすいポイントまでを一気に扱います。

MongoDB 7.x系のmongoshを前提に進めます。

集約パイプラインの基本構造

パイプラインステージという考え方

SQLのSELECTは基本的に1つの文の中に条件・集計・並び替えをまとめて書きますが、集約パイプラインは処理を「ステージ」に分割し、配列として順番に並べます。

各ステージはドキュメントの集合を受け取り、加工した結果を次のステージに渡します。

Unixパイプの|をイメージすると分かりやすいです。

// 基本形：db.コレクション名.aggregate([ステージ1, ステージ2, ...])
db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } }
]);

このパイプラインは「発送済みの注文だけ絞り込む」→「顧客ごとに金額を合計する」→「合計金額の降順に並べる」という3段階の処理を、上から順に実行します。

mongoshでの基本構文

集約パイプラインの結果はカーソルとして返るため、find()と同様に.toArray()やforループで取り出せます。

// 結果を配列として受け取る
const result = db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
]).toArray();

printjson(result);

ステージの順序を変えると結果も処理速度も変わる点が、SQLとの大きな違いです。

$matchはできる限り早い段階に置くのが鉄則で、これによって後続のステージが処理するドキュメント数を減らせます。

主要ステージの使い方

match・group・$sortの組み合わせ

実務で最も使う頻度が高いのが、この3ステージの組み合わせです。

// カテゴリごとの売上件数と平均単価を集計する
db.sales.aggregate([
  { $match: { saleDate: { $gte: ISODate("2026-01-01") } } },
  { $group: {
      _id: "$category",
      count: { $sum: 1 },
      avgPrice: { $avg: "$price" }
  }},
  { $sort: { count: -1 } }
]);

$groupの_idにはグルーピングのキーを指定し、$sum・$avg・$max・$minといった集約演算子をフィールドごとに組み合わせます。

_id: nullを指定すると、コレクション全体を1つのグループとして集計できます。

// 全体の合計金額を1件のドキュメントで取得する
db.sales.aggregate([
  { $group: { _id: null, totalRevenue: { $sum: "$price" } } }
]);

$projectでの出力整形

$projectは出力するフィールドを選択・加工するステージで、SQLのSELECTのフィールド指定に近い役割です。

// 必要なフィールドだけを取り出し、税込み価格を計算して追加する
db.sales.aggregate([
  { $match: { category: "electronics" } },
  { $project: {
      _id: 0,
      name: 1,
      priceWithTax: { $multiply: ["$price", 1.1] }
  }}
]);

_id: 0を指定しない限り、_idフィールドは自動的に出力に含まれる点は覚えておくとよいです。

よくあるつまずきポイント・エラー対処

ステージの順序ミスによるメモリエラー

集約パイプラインには、1ステージあたり100MBのメモリ上限（allowDiskUseオプションを使わない場合の既定値）があります。

これを超えると、次のようなエラーに遭遇します。

MongoServerError: Sort exceeded memory limit of 104857600 bytes,
but did not opt in to external sorting.

筆者が実際に、注文履歴30万件を$sortしてから$matchで絞り込むパイプラインを組んだ際に、このエラーに遭遇したことがあります。

原因は$matchを先に置かず、全件を$sortした後に絞り込んでいたことでした。

❌ Before（絞り込む前に全件ソートしてメモリ上限に達する）

db.orders.aggregate([
  { $sort: { amount: -1 } },
  { $match: { status: "shipped" } },
  { $limit: 10 }
]);

✅ After（先に$matchで絞り込んでからソートする）

db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $sort: { amount: -1 } },
  { $limit: 10 }
]);

$matchを先頭に移動し、さらにamountフィールドにインデックスを張ったことで、このパイプラインは処理対象のドキュメント数が大幅に減り、エラーが解消しました。

どうしても大量データを扱う必要がある場合は、{ allowDiskUse: true }をオプションに追加すればディスクを使った処理に切り替えられますが、速度は落ちるため、まずはステージの順序とインデックスの見直しを優先すべきです。

// ディスクを使った処理を許可する（速度は低下する）
db.orders.aggregate(
  [{ $sort: { amount: -1 } }],
  { allowDiskUse: true }
);

応用・一歩先の使い方

$lookupによるコレクション間の結合

集約パイプラインには$lookupというステージがあり、SQLのJOINに近いことができますが、性質はかなり異なります。

// 注文コレクションに顧客コレクションの情報を結合する
db.orders.aggregate([
  { $match: { status: "shipped" } },
  { $lookup: {
      from: "customers",
      localField: "customerId",
      foreignField: "_id",
      as: "customerInfo"
  }},
  { $unwind: "$customerInfo" }
]);

$lookupは常に左外部結合（LEFT OUTER JOIN相当）として動作し、一致するドキュメントは配列としてasに指定したフィールドに格納されます。

そのため、配列を展開して1件ずつのドキュメントに戻す$unwindステージとセットで使うのが定番のパターンです。

複数コレクションを何段も$lookupでつなぐと、パフォーマンスが大きく劣化しやすいため、リレーションを多用する設計であれば、そもそもRDBMSを検討すべきというのが実務での判断です（この判断軸はMongoDB編最終回で詳しく比較します）。

まとめ

この記事のポイント

  • 集約パイプラインはステージを配列で並べ、ドキュメントを段階的に加工・集計する仕組みである
  • $matchはできる限り早い段階に置き、処理対象のドキュメント数を減らすのが鉄則
  • 1ステージあたり100MBのメモリ上限があり、超えるとエラーになる。allowDiskUseは最終手段
  • $lookupはSQLのJOINとは性質が異なり、多用するとパフォーマンスが劣化しやすい

次に読むべき記事

  • インデックスの基本とパフォーマンス（MongoDB編7本目）
  • スキーマ設計の考え方：埋め込みvs参照（MongoDB編8本目）
  • MongoDBとPostgreSQL（JSONB）比較検証（MongoDB編最終回）

タグ: MongoDB, 上級者向け, 集計

タイトルとURLをコピーしました