MongoDB 聚合管道高级用法

MongoDB 聚合管道高级用法

MongoDB 的聚合管道（Aggregation Pipeline）是处理复杂数据分析和转换的强大工具。它通过一系列阶段（Stage）对文档进行筛选、转换、分组、排序等操作，最终输出处理结果。聚合管道支持多种操作符，可以实现类似 SQL 的 GROUP BY、JOIN、WHERE 等功能，且语法更加声明式和灵活。

聚合管道基础

db.collection.aggregate([
  { $match: { status: "active" } },    // 筛选
  { $group: { _id: "$category", total: { $sum: "$amount" } } },  // 分组
  { $sort: { total: -1 } },            // 排序
  { $limit: 10 },                       // 限制数量
]);

常用阶段操作符

$match：筛选文档，支持索引优化

db.orders.aggregate([
  {
    $match: {
      createdAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2025-01-01") },
      status: "completed",
      totalAmount: { $gt: 100 }
    }
  }
]);

$group：分组聚合

db.orders.aggregate([
  {
    $group: {
      _id: "$customerId",
      totalSpent: { $sum: "$totalAmount" },
      orderCount: { $sum: 1 },
      avgOrderValue: { $avg: "$totalAmount" },
      maxOrder: { $max: "$totalAmount" },
      minOrder: { $min: "$totalAmount" },
    }
  }
]);

$project：字段选择和计算

db.orders.aggregate([
  {
    $project: {
      orderId: "$_id",
      customerName: "$customer.name",
      totalWithTax: { $multiply: ["$totalAmount", 1.06] },
      year: { $year: "$createdAt" },
      month: { $month: "$createdAt" },
    }
  }
]);

$lookup：关联查询（类似 LEFT JOIN）

db.orders.aggregate([
  {
    $lookup: {
      from: "customers",
      localField: "customerId",
      foreignField: "_id",
      as: "customerInfo"
    }
  },
  { $unwind: "$customerInfo" },
  {
    $project: {
      orderId: "$_id",
      customerName: "$customerInfo.name",
      totalAmount: 1,
    }
  }
]);

条件聚合

$cond：条件表达式

db.orders.aggregate([
  {
    $project: {
      orderId: "$_id",
      discount: {
        $cond: [
          { $gte: ["$totalAmount", 1000] },  // 条件
          0.1,                              // 条件为 true
          0                                 // 条件为 false
        ]
      }
    }
  }
]);

$switch：多条件分支

db.orders.aggregate([
  {
    $project: {
      category: {
        $switch: {
          branches: [
            { case: { $gte: ["$totalAmount", 1000] }, then: "high-value" },
            { case: { $gte: ["$totalAmount", 500] }, then: "medium-value" },
          ],
          default: "low-value"
        }
      }
    }
  }
]);

数组操作

$unwind：展开数组为多个文档

db.orders.aggregate([
  { $unwind: "$items" },
  {
    $group: {
      _id: "$items.productId",
      totalSold: { $sum: "$items.quantity" },
    }
  }
]);

$filter：数组过滤

db.orders.aggregate([
  {
    $project: {
      orderId: "$_id",
      expensiveItems: {
        $filter: {
          input: "$items",
          as: "item",
          cond: { $gte: ["$$item.price", 100] }
        }
      }
    }
  }
]);

$map：数组映射

db.orders.aggregate([
  {
    $project: {
      itemPrices: {
        $map: {
          input: "$items",
          as: "item",
          in: "$$item.price"
        }
      }
    }
  }
]);

日期操作

db.orders.aggregate([
  {
    $project: {
      year: { $year: "$createdAt" },
      month: { $month: "$createdAt" },
      dayOfWeek: { $dayOfWeek: "$createdAt" },
      hour: { $hour: "$createdAt" },
      isWeekend: {
        $in: [{ $dayOfWeek: "$createdAt" }, [1, 7]]
      }
    }
  }
]);

注意事项

• 聚合管道中，越早使用 $match 越好，可以利用索引减少后续阶段处理的数据量
• $group 是无序操作，后续如需有序，应在 $group 之后加 $sort
• $lookup 的 from 必须是同一数据库中的集合
• 聚合管道有内存限制（默认 100MB），大量数据处理可使用 allowDiskUse 选项
• 复杂的聚合操作应先在测试环境验证结果正确性

MongoDB 聚合管道是处理复杂数据分析的利器，掌握各种操作符的组合使用，能够在数据库层面完成大多数数据处理任务，减少应用层代码复杂度。