聚合管道进阶

约 537 字大约 2 分钟

布欧-Lewyon

2026-05-15

$unwind

将数组字段展开为多个文档：

// 原始文档
db.orders.findOne()
// { _id: 1, items: [{ product: "A", qty: 2 }, { product: "B", qty: 1 }] }

// 展开
db.orders.aggregate([
  { $unwind: "$items" }
])
// { _id: 1, items: { product: "A", qty: 2 } }
// { _id: 1, items: { product: "B", qty: 1 } }

// 保留空数组的文档
db.orders.aggregate([
  { $unwind: { path: "$items", preserveNullAndEmptyArrays: true } }
])

$lookup

左外连接（类似 SQL 的 LEFT JOIN）：

// orders 集合中有 userId，需要关联 users 集合
db.orders.aggregate([
  { $lookup: {
    from: "users",                  // 关联的集合
    localField: "userId",           // 当前集合字段
    foreignField: "_id",            // 目标集合字段
    as: "user"                      // 输出数组字段名
  }},
  { $unwind: "$user" },            // 展开为对象（可选）
  { $project: {
    _id: 1,
    total: 1,
    "user.name": 1,
    "user.email": 1
  }}
])

// 关联 + 管道子查询
db.orders.aggregate([
  { $lookup: {
    from: "reviews",
    let: { productId: "$productId" },
    pipeline: [
      { $match: { $expr: { $eq: ["$productId", "$$productId"] } } },
      { $sort: { createdAt: -1 } },
      { $limit: 3 }
    ],
    as: "recentReviews"
  }}
])

$addFields / $set

添加新字段：

db.orders.aggregate([
  { $addFields: {
    tax: { $multiply: ["$amount", 0.1] },
    totalWithTax: { $add: ["$amount", { $multiply: ["$amount", 0.1] }] },
    year: { $year: "$createdAt" }
  }}
])

$bucket

自动分桶统计：

db.orders.aggregate([
  { $bucket: {
    groupBy: "$amount",
    boundaries: [0, 100, 500, 1000, 10000],
    default: "Other",
    output: {
      count: { $sum: 1 },
      total: { $sum: "$amount" }
    }
  }}
])
// { _id: 0, count: 25, total: 1250 }
// { _id: 100, count: 40, total: 8600 }
// ...

多维度聚合（多管道并行）：

db.orders.aggregate([
  { $facet: {
    byCategory: [
      { $group: { _id: "$category", count: { $sum: 1 }, total: { $sum: "$amount" } } }
    ],
    byStatus: [
      { $group: { _id: "$status", count: { $sum: 1 } } }
    ],
    stats: [
      { $group: { _id: null, avgAmount: { $avg: "$amount" }, totalRevenue: { $sum: "$amount" } } }
    ]
  }}
])
// 输出三个独立的分组结果

完整分析示例

db.orders.aggregate([
  // 第一步：筛选
  { $match: { status: "completed", createdAt: { $gte: ISODate("2026-01-01") } } },

  // 第二步：展开商品明细
  { $unwind: "$items" },

  // 第三步：按分类聚合
  { $group: {
    _id: "$items.category",
    totalSales: { $sum: { $multiply: ["$items.price", "$items.qty"] } },
    orderCount: { $sum: 1 },
    avgItemPrice: { $avg: "$items.price" }
  }},

  // 第四步：排序
  { $sort: { totalSales: -1 } },

  // 第五步：关联分类名称
  { $lookup: {
    from: "categories",
    localField: "_id",
    foreignField: "code",
    as: "category"
  }},

  // 第六步：投射
  { $project: {
    categoryName: { $arrayElemAt: ["$category.name", 0] },
    totalSales: 1,
    orderCount: 1,
    avgItemPrice: { $round: ["$avgItemPrice", 2] }
  }}
])

小结

$unwind 展开数组字段；$lookup 跨集合关联。
$addFields / $set 添加计算字段。
$bucket 分桶统计；$facet 多维度并行聚合。
聚合管道是 MongoDB 数据处理的精华，支持多阶段组合。
每个阶段的输出作为下一阶段的输入。