【ITニュース解説】Mastering MongoDB Aggregation Pipelines: A Developer’s Complete Guide

MongoDBのAggregation Pipelinesは、データを効率的に処理するための強力なツールだ。SQLの知識がある人にとっては、MongoDBでも同様の操作ができると理解しやすいだろう。この記事では、Aggregation Pipelinesの基本的な概念から、実際のコード例を用いた解説までを提供する。

Aggregation Pipelineは、データ処理を段階的に行うプロセスだ。各段階（ステージ）は、ドキュメントを受け取り、処理を行い、その結果を次のステージに渡す。これは、複数のフィルターが連なったチェーンのようなものだと考えるとわかりやすい。例えば、「ユーザーをフィルタリングする」→「購読情報を結合する」→「新しいフィールドを追加する」→「必要なフィールドのみを選択する」という流れでデータ処理を行うことができる。

基本的な構造は以下の通りだ。

Copy
1db.collection.aggregate([
2  { stage1 },
3  { stage2 },
4  { stage3 }
5])

各 {} がステージを表し、ステージの順序が重要となる。前のステージの出力が、次のステージの入力となるからだ。最初のステージで処理するデータ量を調整することで、後続のステージの効率を上げることが可能だ。

MongoDBでは、ステージや演算子は常に $ で始まる。これはMongoDBに対して、「これは特別なコマンドであり、フィールド名ではない」ということを示すためのものだ。例えば、$match はSQLの WHERE 句に相当し、$group は GROUP BY 句に相当する。$project は SELECT 句、$lookup は JOIN 句に相当すると考えると理解しやすい。

以下は、主要なAggregation StagesとSQLの対応を示す表だ。

MongoDB Stage	SQL Equivalent	Example
$match	WHERE	年齢が20歳を超えるユーザーをフィルタリングする
$project	SELECT col1, col2	名前とメールアドレスのみを表示する
$group	GROUP BY	国ごとのユーザー数をカウントする
$sort	ORDER BY	ユーザーを参加日でソートする
$limit	LIMIT	上位10人のユーザーのみを表示する
$lookup	JOIN	ユーザーと購読情報を結合する

実際のプロジェクトで使用されているコード例を見てみよう。これは、ユーザーのチャンネルプロファイル（YouTubeのようなもの）を取得するコントローラーの例だ。

Copy
1const getUserChannelProfile = asyncHandler(async (req, res) => {
2  const { username } = req.params;
3  if (!username?.trim()) throw new ApiError(400, "Channel id is required");
4
5  const channel = await User.aggregate([
6    { $match: { username: username?.toLowerCase() } },
7    {
8      $lookup: {
9        from: "Subscription",
10        localField: "_id",
11        foreignField: "channel",
12        as: "subscribers",
13      },
14    },
15    {
16      $lookup: {
17        from: "Subscription",
18        localField: "_id",
19        foreignField: "subscriber",
20        as: "subscribedTo",
21      },
22    },
23    {
24      $addFields: {
25        subscribersCount: { $size: "$subscribers" },
26        channelsSubscribedToCount: { $size: "$subscribedTo" },
27        issSubscribed: {
28          $cond: {
29            if: { $in: [req.user?._id, "$subscribers.subscriber"] },
30            then: true,
31            else: false,
32          },
33        },
34      },
35    },
36    {
37      $project: {
38        fullname: 1,
39        username: 1,
40        avatar: 1,
41        coverImage: 1,
42        subscribersCount: 1,
43        channelsSubscribedToCount: 1,
44        issSubscribed: 1,
45        email: 1,
46        createdAt: 1,
47      },
48    },
49  ]);
50
51  if (!channel || channel?.length === 0)
52    throw new ApiError(404, "Channel not found");
53
54  return res
55    .status(200)
56    .json(new ApiResponse(200, "Channel fetched successfully", channel[0]));
57});

このコードは、以下のステップで動作する。

1. 入力チェック: リクエストにユーザー名が含まれているかを確認する。
2. $match: username に一致するユーザーをフィルタリングする（SQLの WHERE username = 'someuser' に相当）。
3. $lookup (Subscribers): Subscription コレクションと結合し、ユーザーのチャンネルを購読しているユーザーを取得する（SQLの JOIN に相当）。
4. $lookup (Subscribed Channels): ユーザーが購読しているチャンネルを取得する。
5. $addFields: 新しいフィールド（購読者数、購読チャンネル数、現在のユーザーが購読しているかどうか）を計算して追加する。
6. $project: 必要なフィールドのみを選択する（SQLの SELECT に相当）。

この例からわかるように、Aggregation Pipelinesを使用すると、複雑なデータ処理を効率的に行うことができる。

さらに、MongoDB Compassを使用すると、Aggregation Pipelinesを視覚的に構築し、各ステージの結果を確認しながらデバッグすることができる。MongoDBには、練習用の無料サンプルデータセット（映画、Airbnbなど）も用意されているので、これらを利用して実際に手を動かしてみると良いだろう。

より高度な演算子として、$unwind（配列フィールドを分解）、$facet（複数のパイプラインを同時に実行）、$bucket（データを範囲ごとにグループ化）、$bucketAuto（自動でバケット境界を決定）、$graphLookup（階層的なデータを検索）、$merge（パイプラインの結果をコレクションに書き込む）などがある。これらの演算子を使いこなせるようになると、さらに複雑なデータ処理が可能になる。

Aggregation Pipelinesは、MongoDBの基本的なCRUD操作だけでなく、SQLと同様のデータ操作を可能にする強力なツールだ。$match でフィルタリングし、$lookup で結合し、$addFields で計算し、$project で選択するというステップを意識することで、効率的なデータ処理を行うことができるようになるだろう。