Posted inデータベース

MongoDB Aggregation Pipeline：複雑なデータの処理・グループ化・変換

Database tutorial - IT technology blog
Database tutorial - IT technology blog

6ヶ月間、2つの本番環境での運用経験があります。月間約5万件の注文を処理するECプラットフォームと、ユーザーイベントデータを分析するダッシュボードです。この稼働実績があれば、Aggregation Pipelineが真価を発揮するタイミングと、デバッグの悪夢になるタイミングが正確に分かります。

複数のプロジェクトでMySQL、PostgreSQL、MongoDBを使ってきましたが、それぞれに適した用途があります。ただし、MongoDBのAggregation Pipelineは、リレーショナルデータベースにはないデータベース内変換の仕組みを持っています。ドキュメントの説明ではなく、実際のところをお話しします。

MongoDBで複雑なデータをクエリする2つの方法

MongoDBを使い始めた頃は、find()で十分に感じます。高速で可読性が高く、ユースケースの80%をカバーできます。しかし残りの20%—合計のグループ化、平均の計算、コレクションの結合、ドキュメントの再構成—になると、find()はすぐに限界を迎えます。

実際には、3つの選択肢があります：

方法1：データをアプリケーションコードに取り込んで処理する。生のドキュメントを取得し、PythonやNode.jsでループ処理して手動で集計します。小さなデータセットでは機能しますが、スケールすると深刻な問題が起きます—単一の数値を計算するためにギガバイト単位のデータをネットワーク経由で転送することになります。

方法2：MapReduce。MongoDBの旧来の集計アプローチです。まだサポートされていますが、実質的には非推奨—冗長で、デバッグが難しく、大規模データセットでは遅い。2021年にあるプロジェクトで使いましたが、2週間後悔し続けました。

方法3：Aggregation Pipeline。ドキュメントをステップバイステップで変換するステージのシーケンスです。各ステージが入力を受け取り、処理して、次のステージに出力を渡します。これが現在の標準で、MongoDBが毎リリースで積極的に最適化しているものです。

比較になりません。私たちのアナリティクスダッシュボードでは、ロジックをパイプラインに移した結果、特定のレポートがネットワーク経由で約800MBの生イベントデータを取得していたものが、2KBの結果セットを返すだけになりました。アプリケーション側の処理では、そのスケールには到底太刀打ちできません。

本番環境6ヶ月後のメリットとデメリット

うまく機能する点

ステージベースのアーキテクチャにより、複雑なクエリが実際にデバッグ可能になります。パイプラインを任意のステージまで実行して中間結果を確認できます—これは単一のSQLクエリやアプリケーション側の処理ループでは絶対にできないことです。何か問題があれば、どこを見ればいいかが正確に分かります。

インデックスの利用は確実です。パイプラインの早い段階に置かれた$matchステージは既存のインデックスを使用できます。つまり、インデックス付きフィールドに対して適切に構成されたパイプラインは、同等のSQLクエリに匹敵するパフォーマンスを発揮します。

柔軟性も本物です。ドキュメントの再構成、派生フィールドの計算、配列の展開、$lookupによるコレクションの結合、ネストされたオブジェクトの処理—すべて1つのクエリで実現できます。PostgreSQLならCTEやウィンドウ関数を使うところですが、MongoDBのパイプラインは多くの実際のスキーマに対してドキュメント形状のデータをより表現力豊かに扱います。

うまく機能しない点

構文が冗長です。適度に複雑な集計は、同等のSQLよりも通常2〜3倍の行数になります。コードベースで15ステージのパイプラインを初めて見たとき、データフローを理解するだけで3回読み返す必要がありました。

メモリ制限に引っかかります。デフォルトでは、各パイプラインステージはRAMが100MBに制限されています。大規模データセットにはallowDiskUse: trueが必要で、実際のパフォーマンスに影響します—深く考えずにオンにできる設定ではありません。

時間的プレッシャーの下でマルチステージパイプラインをデバッグするのは苦痛です。パイプラインが予期しない結果を出したとき、10以上のステージを追跡するには相当な労力が必要です。最終的に$addFieldsを主なデバッグツールとして採用しました—中間値を検査するための一時フィールドを追加し、リリース前に削除するという方法です。

推奨セットアップ

集計クエリを書く前に、いくつかのセットアップの決定が重要です：

$matchフィールドには先にインデックスを付ける。インデックスのないフィールドで$matchから始まるAggregation Pipelineは、コレクション全体をスキャンします。statusフィールドとdateフィールドに複合インデックスを追加するまで、アナリティクスクエリのパフォーマンスが壊滅的でした。

$matchと$sortはできるだけ早い段階に置く。これらのステージはインデックスの恩恵を受けます。それらの前に変換ステージがあると、それらの操作のインデックス最適化が無効になります。

新しい集計をデプロイする前にexplain()を使う。MongoDBのexplain("executionStats")は、パイプラインステージがインデックスを使用しているかCOLLSCANにフォールバックしているかを示します。新しいものをリリースする前に実行してください—30秒かかるだけで、スケール時に問題になっていたケースを何度も発見できました。

db.orders.explain("executionStats").aggregate([
  { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
])

大規模な結果セットを意図的に処理するアナリティクスクエリにのみallowDiskUse: trueを設定する。OLTPクエリでこれが必要な場合は、ほぼ確実にインデックスの欠如またはパイプラインの構造的な問題のサインです。

実装ガイド

コアステージの種類

6つのステージが実世界のユースケースの90%をカバーします。他のものに手を出す前に、これらを深く学んでください：

  • $match — ドキュメントをフィルタリング（SQL の WHERE に相当）
  • $group — ドキュメントをグループ化して集計を計算（SQL の GROUP BY に相当）
  • $project — ドキュメントを再構成し、フィールドを含めるか除外する
  • $sort — 結果を並べ替える
  • $limit / $skip — ページネーション
  • $lookup — 別のコレクションを結合（SQL の LEFT JOIN に相当）

グループ化と集計の計算

最もよくあるのは、グループごとの合計、件数、または平均が欲しい場合です。パターンはこうです：

// 売上合計上位10顧客
db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $group: {
      _id: "$customerId",
      totalRevenue: { $sum: "$amount" },
      orderCount: { $sum: 1 },           // MongoDB 5.0以降は { $count: {} } を使用
      avgOrderValue: { $avg: "$amount" },
      lastOrder: { $max: "$createdAt" }
    }
  },
  { $sort: { totalRevenue: -1 } },
  { $limit: 10 }
])

MongoDBはデータがフラットでない場合、SQLよりもネストされたドキュメントフィールドや配列を自然に扱います—ドキュメント指向スキーマの真の利点です。

$unwind を使った配列の処理

MongoDBの集計がSQLを明確に上回る点の一つが配列データの処理です。$unwindステージは配列フィールドを個別のドキュメントに展開し、要素ごとの集計を可能にします。

// 各商品を含む注文数を集計
db.orders.aggregate([
  { $unwind: "$items" },
  {
    $group: {
      _id: "$items.productId",
      orderCount: { $sum: 1 },
      totalQuantity: { $sum: "$items.quantity" }
    }
  },
  { $sort: { orderCount: -1 } }
])

$unwindなしでは、配列を反復処理するためにコレクション全体をアプリケーションメモリに引き込む必要があります。平均5〜10アイテムを持つ100万件の注文コレクションでは、10msの集計と数秒のデータ転送の差になります。

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

$lookupはMongoDBがJOINを処理する方法です。SQLより冗長ですが、一対多の関係に対して効果的です：

// 注文に顧客詳細情報を付加
db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $lookup: {
      from: "customers",
      localField: "customerId",
      foreignField: "_id",
      as: "customerDetails"
    }
  },
  { $unwind: "$customerDetails" },
  {
    $project: {
      orderId: "$_id",
      amount: 1,
      customerName: "$customerDetails.name",
      customerEmail: "$customerDetails.email"
    }
  }
])

本番環境で遭遇した注意点：外部コレクションのインデックスに対する$lookupの動作はMongoDBのバージョンによって異なります。複雑な結合でより良いインデックス制御を行うには、$lookupのパイプライン形式（MongoDB 3.6以降で利用可能）を使用してください。

実際のパイプライン：月次売上レポート

これは時系列レポートのために定期的に使用するパターンです：

db.orders.aggregate([
  // ステップ1：対象日付範囲でフィルタリング
  {
    $match: {
      createdAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2025-01-01") },
      status: { $in: ["completed", "shipped"] }
    }
  },
  // ステップ2：日付から月を抽出
  {
    $addFields: {
      month: { $month: "$createdAt" },
      year: { $year: "$createdAt" }
    }
  },
  // ステップ3：年月でグループ化
  {
    $group: {
      _id: { year: "$year", month: "$month" },
      revenue: { $sum: "$amount" },
      orders: { $sum: 1 },
      uniqueCustomers: { $addToSet: "$customerId" }
    }
  },
  // ステップ4：セットからユニーク顧客数を計算
  {
    $addFields: {
      uniqueCustomerCount: { $size: "$uniqueCustomers" }
    }
  },
  { $project: { uniqueCustomers: 0 } },
  { $sort: { "_id.year": 1, "_id.month": 1 } }
])

PostgreSQLの同等処理にはCTEと少なくとも1つのウィンドウ関数が必要です。MongoDBのパイプラインは論理的なステップに直接マッピングされます—各ステージが単一の明確な役割を持ち、冗長な構文にもかかわらずレビューを容易にします。

PyMongo を使った Python での利用

私の本番環境での集計のほとんどはPyMongoを通じて実行しており、パイプラインは単なるdictのリストです—ユーザーフィルターに基づいて動的に構築しやすい：

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["ecommerce"]

pipeline = [
    {"$match": {"status": "completed"}},
    {"$group": {
        "_id": "$customerId",
        "total": {"$sum": "$amount"},
        "count": {"$sum": 1}   # MongoDB 5.0以降は {"$count": {}} を使用
    }},
    {"$sort": {"total": -1}},
    {"$limit": 20}
]

results = list(db.orders.aggregate(pipeline, allowDiskUse=True))
for r in results:
    print(f"顧客 {r['_id']}: ${r['total']:.2f}（{r['count']} 件の注文）")

パイプラインをPythonのデータ構造として構築することで、個々のステージを独立してユニットテストできます。それだけで過去6ヶ月間のCIデバッグを数時間節約できました—生のSQL文字列では絶対にできないことです。

まとめ

Aggregation Pipelineは、MongoDB内の変換作業に適したツールです。6ヶ月後の私の核心的な推奨はこれです：ODMヘルパーや抽象化レイヤーに手を出す前に、6つの基本ステージを完全に習得してください。ヘルパーは、何か壊れたときに理解する必要があるものをちょうど抽象化してしまいます。

私がデバッグした集計バグのほとんどは、2つのパターンに起因します。1つ目は、$matchの前に$groupを追加すること—インデックスの使用を無効にし、10msのクエリを数秒のフルスキャンに変えます。2つ目は、最初にunwindせずに配列フィールドで集計すること—クロスチェックするまで妥当に見える形でカウントが膨れ上がります。

SQLから来た方へ：各ステージをクエリ句にマッピングしてください：$match → WHERE、$group → GROUP BY、$sort + $limit → ORDER BY + LIMIT。冗長さは本物で、好きになることはないでしょう。しかしネストされた配列と動的スキーマを持つドキュメント形状のデータに対しては、パイプラインのデバッグ可能性と柔軟性が正しいトレードオフです。

Share:
Tags: