MongoDB 聚合管道

前面学的 find 只能做"过滤 + 投影",遇到分组统计、字段变形、关联另一集合这类需求就无能为力了。MongoDB 给的答案是 聚合管道(aggregation pipeline)——把数据想象成水流,依次穿过多个"阶段"(stage),每个阶段做一种处理,最终输出结果。功能上相当于 SQL 的 GROUP BY + JOIN + 子查询的合体。

1. 管道是什么:数据流水线

聚合的核心思想是流水线:一条文档依次穿过 $match → $sort → $limit 等阶段,每个阶段处理完输出给下一个。所有阶段写在一个数组里,顺序就是执行顺序。

// 聚合管道(pipeline):数据像水流依次穿过多个 stage
// 每个 stage 用一个对象表示,前一个的输出是后一个的输入
// 写法: db.collection.aggregate([ stage1, stage2, ... ])

// 一个最简单的例子:筛选 -> 排序 -> 取前 3
db.posts.aggregate([
  { $match: { views: { $gte: 10 } } },   // stage 1: 过滤
  { $sort: { views: -1 } },              // stage 2: 排序
  { $limit: 3 }                          // stage 3: 取前 3
])

// 这其实和 find().sort().limit() 等价
// 但聚合的能力远不止于此 -- 它能做 find 做不到的"分组、变形、关联"

和 find 最大的区别是:聚合能改变文档的形状、做分组统计、关联其他集合——这些 find 一概做不了。

2. $match 与 $group:分组统计的灵魂

$match 过滤(尽量放最前减少后续处理量),$group 按字段分组并做聚合运算。两者组合是聚合最经典的用法。

// 经典分组统计:按作者统计总阅读量和文章数
db.posts.aggregate([
  // 1. $match 先过滤掉不要的数据(尽早过滤减少后续计算)
  { $match: { status: "published" } },

  // 2. $group 按 author 字段分组,组内做聚合
  {
    $group: {
      _id: "$author",                    // 按 author 分组("$" 表示引用字段值)
      totalViews: { $sum: "$views" },    // 组内 views 求和
      count: { $sum: 1 },                // 组内文档数(每条 +1)
      avgViews: { $avg: "$views" },      // 平均阅读量
      maxView: { $max: "$views" },       // 最大阅读量
      minView: { $min: "$views" },       // 最小阅读量
      titles: { $push: "$title" }        // 把所有 title 收集成数组
    }
  },

  // 3. 按总阅读量倒序
  { $sort: { totalViews: -1 } },

  // 4. 只取前 10 名
  { $limit: 10 }
])

// $group 的 _id 还可以按多字段分组(传一个对象)
{
  $group: {
    _id: { author: "$author", status: "$status" },   // 作者+状态组合分组
    count: { $sum: 1 }
  }
}
// 想不分组只算总数:_id: null

几个要点:

3. $project 与 $addFields:字段重塑

$project 类似 SQL 的 SELECT——选哪些字段、重命名、生成计算字段。$addFields 则是在原文档基础上加字段,不动其他字段。

// $project:重塑每条文档的形状(类似 SQL 的 SELECT)
db.posts.aggregate([
  {
    $project: {
      _id: 0,                        // 不要 _id
      title: 1,                      // 保留 title
      views: 1,                      // 保留 views
      authorName: "$author",         // 把 author 字段重命名为 authorName
      // 计算字段:用表达式生成新字段
      isPopular: { $gte: ["$views", 100] },     // 阅读量 >= 100 -> true
      summary: { $concat: ["$title", " (", { $toString: "$views" }, ")"] }
    }
  }
])

// $addFields:在原文档基础上加字段,不删除其他字段(比 $project 更省事)
db.posts.aggregate([
  { $addFields: {
      isPopular: { $gte: ["$views", 100] }
  }}
])

// $unset:删除字段(等价于 $project 排除)
db.posts.aggregate([{ $unset: "content" }])

// 常用表达式操作符:
// 算术: $add $subtract $multiply $divide $mod
// 字符串: $concat $toUpper $toLower $substr $split
// 日期: $year $month $dayOfMonth $hour $dateToString
// 比较: $eq $ne $gt $lt $gte $lte $cmp
// 逻辑: $and $or $not $cond $switch $ifNull

表达式操作符是聚合的"瑞士军刀":算术($add / $multiply)、字符串($concat / $toUpper)、日期($year / $dateToString)、条件($cond / $switch / $ifNull)。它们让你在管道里做几乎任何数据变换。

4. $unwind:展开数组

$unwind 把数组字段拆成多条文档,每条对应数组里的一个元素。配合 $group 能做经典的"热门标签统计"。

// $unwind:把数组字段"展开"成多条文档,每条一个元素
// 原文档: { _id: 1, title: "A", tags: ["入门", "笔记"] }
// unwind 后变成两条:
//   { _id: 1, title: "A", tags: "入门" }
//   { _id: 1, title: "A", tags: "笔记" }

db.posts.aggregate([
  { $unwind: "$tags" },            // 每个 tag 一条
  { $group: {
      _id: "$tags",                // 按 tag 分组
      count: { $sum: 1 }           // 每个 tag 出现多少次
  }},
  { $sort: { count: -1 } }
])
// 这就是经典的 "热门标签榜" 实现

// preserveNullAndEmptyArrays:数组为空时是否保留文档
{ $unwind: { path: "$tags", preserveNullAndEmptyArrays: true } }

这是聚合里非常常用的"展开 → 分组"模式。注意 $unwind 会放大文档数(一个含 3 个元素的数组会变成 3 条文档),数据量大时要谨慎,先 $match 过滤。

5. $lookup:关联另一个集合

MongoDB 不像关系数据库那样天生支持 JOIN,但 $lookup 阶段能在聚合管道里实现左连接——把另一个集合里匹配的文档"塞"进当前文档的数组字段。

// $lookup:左连接另一个集合(类似 SQL 的 LEFT JOIN)
// 把另一个集合里匹配的文档"塞"进当前文档的一个数组字段
db.posts.aggregate([
  {
    $lookup: {
      from: "authors",             // 要关联的集合
      localField: "authorId",      // 当前文档的字段
      foreignField: "_id",         // 目标集合的匹配字段
      as: "authorInfo"             // 结果放进这个数组字段
    }
  }
])
// 结果:每条 post 文档多了 authorInfo 数组(里面是匹配到的 author 文档)

// $unwind + $lookup:把结果从数组展开成单个对象
db.posts.aggregate([
  { $lookup: { from: "authors", localField: "authorId",
               foreignField: "_id", as: "authorInfo" } },
  { $unwind: "$authorInfo" }       // 拆掉数组,变成对象
])

// 高级版:用 pipeline 形态做更复杂的关联
db.posts.aggregate([
  { $lookup: {
      from: "comments",
      let: { postId: "$_id" },     // 把当前文档字段传给子管道
      pipeline: [
        { $match: { $expr: { $eq: ["$postId", "$$postId"] } } },
        { $match: { visible: true } },
        { $count: "commentCount" }
      ],
      as: "commentStats"
  }}
])

$lookup 的结果始终是数组(即使只匹配到一条)。配合 $unwind 可以把它变成单个对象。如果关联条件复杂(比如还要在右表加额外过滤),用 pipeline 形态配 $expr。

6. $bucket 与 $facet:高级统计

$bucket 按区间分桶(常用于直方图、年龄段统计),$facet 则能在一次请求里并行跑多条独立管道——非常适合做仪表盘那种"一次拿多张报表"。

// $bucket:按区间把文档"分桶"(常用于做直方图、年龄段统计)
db.users.aggregate([
  { $bucket: {
      groupBy: "$age",                      // 按年龄分桶
      boundaries: [0, 18, 30, 45, 60, 100], // 桶边界
      default: "other",                     // 落在所有边界之外的归这里
      output: {                             // 每个桶统计什么
        count: { $sum: 1 },
        names: { $push: "$name" }
      }
  }}
])
// 输出每个年龄段 [0,18) [18,30) [30,45) ... 的人数

// $facet:同时跑多条独立的聚合管道(一次请求算多张报表)
db.posts.aggregate([
  { $facet: {
      "byYear": [
        { $group: { _id: "$year", count: { $sum: 1 } } }
      ],
      "topViewed": [
        { $sort: { views: -1 } },
        { $limit: 5 }
      ],
      "total": [
        { $count: "n" }
      ]
  }}
])
// 结果: { byYear: [...], topViewed: [...], total: [{n: 100}] }

7. 完整 stage 速查表

8. 性能注意

聚合功能强大但执行成本不低,大型集合上写错聚合能让 CPU 拉满。几条经验法则:

小结

这一篇覆盖了 MongoDB 聚合管道的核心:流水线思想、$match 过滤、$group 分组、$project 重塑、$unwind 展开、$lookup 关联。从此你能用 MongoDB 完成报表、统计、数据清洗等复杂任务。下一篇进入数据库性能的生命线——索引,让查询速度从秒级降到毫秒级。

← 上一篇 MongoDB 更新操作符

下一篇 MongoDB 索引 →

✈️💬