聚合框架（Aggregation Pipeline）

候选人小林在面试字节跳动的数据工程师岗位时，面试官看了看他的简历，问道：

"你在项目中用 MongoDB 聚合管道做过什么复杂的数据分析？"

小林说："做过用户画像的统计，比如按年龄段分组、计算活跃天数什么的。"

面试官点点头，又问："那你有没有遇到过聚合查询很慢的情况？怎么优化的？"

小林说："加索引吧...或者限制返回文档数量？"

面试官追问："那你知不知道聚合管道的阶段顺序会影响性能？比如 $match 和 $sort 谁先执行？"

小林停顿了一下，说："好像是 $match 先？"

面试官继续问："那如果 $match 的条件能过滤掉 99% 的数据，但你的 $match 写在 $sort 后面，MongoDB 会怎么处理？"

小林彻底不知道了。

【面试官心理】这道题我是在测试候选人对聚合管道执行模型的了解。知道聚合管道是什么的占 80%，能说出基本阶段的占 50%，知道阶段顺序优化原则的占 20%，能在生产环境中实际进行优化的不到 10%。聚合管道是 MongoDB 中最强大的工具，也是最容易写出"慢查询"的地方。

一、聚合管道基础 🔴

1.1 什么是聚合管道

聚合管道（Aggregation Pipeline）是 MongoDB 用来处理数据的框架，类似于 Linux 的管道命令。每个阶段（Stage）接收上一阶段的输出，处理后传递给下一阶段：

// Linux 管道命令类比
cat access.log | grep "ERROR" | awk '{print $1}' | sort | uniq -c | sort -rn

// MongoDB 聚合管道（等价逻辑）
db.accessLogs.aggregate([
  { $match: { status: "ERROR" } },      // grep：过滤 ERROR 日志
  { $group: { _id: "$ip", count: { $sum: 1 } } },  // awk+uniq+wc：按 IP 分组计数
  { $sort: { count: -1 } }             // sort：按出现次数降序
])

1.2 核心阶段详解

// $match：过滤文档（尽可能靠前放）
{ $match: { status: "active", age: { $gte: 18 } } }

// $project：选择/重命名字段
{ $project: {
    _id: 0,                          // 排除 _id
    name: 1,                         // 包含 name
    email: { $toLower: "$email" },  // 转换大小写
    fullAddress: {                   // 字段拼接
      $concat: ["$city", "-", "$district", "-", "$detail"]
    }
  }
}

// $group：分组聚合
{ $group: {
    _id: "$category",               // 分组字段
    totalSales: { $sum: "$amount" }, // 求和
    avgPrice: { $avg: "$price" },   // 平均值
    maxQuantity: { $max: "$quantity" }, // 最大值
    products: { $addToSet: "$productName" }, // 去重收集
    orderIds: { $push: "$orderId" }  // 不去重收集
  }
}

// $sort：排序（在内存中执行，超过 100MB 需要设置 allowDiskUse）
{ $sort: { totalSales: -1 } }

// $limit：限制返回数量
{ $limit: 10 }

// $unwind：展开数组（多对多关系的处理神器）
{ $unwind: "$tags" }  // tags: ["java","python"] → 两个文档

1.3 ❌ 错误示范

候选人原话："MongoDB 的聚合管道就是 SQL 的 GROUP BY，功能一样。"

问题诊断：

忽略了聚合管道的阶段化处理模型——SQL 是一次性执行，管道是流式处理
没有理解 $lookup 和 SQL JOIN 的性能差异
混淆了 SQL 的 HAVING 和 $match 的作用时机
不了解聚合管道的内存限制（默认 100MB）

面试官内心 OS："说聚合管道等于 GROUP BY 的候选人，说明他的 MongoDB 经验基本停留在'换个语法的 SQL'层面。聚合管道的强大在于阶段组合、表达式计算和 $lookup，但很多候选人只用过 $match + $group + $sort 这三板斧。"

1.4 标准回答

// P5 级别：知道聚合管道的基本用法
"聚合管道是 MongoDB 的数据处理框架，用方括号括起来的阶段数组表示，
每个阶段是一个文档。每个阶段接收上一阶段的输出。"

// P6 级别：能讲清楚主要阶段和基本优化原则
"主要阶段有 $match、$group、$sort、$project、$unwind、$lookup 等。
优化原则是：$match 和 $limit 尽可能靠前，减少后续阶段处理的数据量。
$sort 必须放在 $limit 之前，否则会先排完所有数据再截断。"

// P7 级别：能讲清楚优化器和 pipeline 优化技巧
"MongoDB 会对管道进行优化，但有限：$match + $sort 可以合并，
$project 可以下推到更早的阶段。但如果 $match 在 $unwind 后面，
MongoDB 无法优化——需要手动调整阶段顺序。
此外，可以用 $facet 做多维度聚合，用 $bucket 做分段统计。"

【面试官心理】聚合管道的追问方向通常是：先问"有哪些阶段"，再问"怎么优化性能"，然后问"$lookup 和 SQL JOIN 有什么区别"，最后问"$facet 和 $bucket 的使用场景"。能答到第三层的 P6 候选人，回答"lookup 就是 JOIN"是扣分项。

二、`$lookup` 与表关联 🟡

2.1 `$lookup` 的基本用法

$lookup 是 MongoDB 3.2 引入的左外连接功能，但它的行为和 SQL JOIN 有显著区别：

// 基础语法：单字段关联
db.orders.aggregate([
  {
    $lookup: {
      from: "products",           // 关联的集合名
      localField: "productId",    // 本集合的字段
      foreignField: "_id",        // 关联集合的字段
      as: "productDetails"       // 输出字段名（数组类型）
    }
  }
])

// 输出示例
{
  _id: 1,
  orderNumber: "ORDER001",
  productId: ObjectId("..."),
  productDetails: [              // $lookup 结果是数组，即使只匹配一条也是数组
    { _id: ObjectId("..."), name: "iPhone 15", price: 6999 }
  ]
}

2.2 `$lookup` 的性能陷阱

这是最容易踩坑的地方：

// ❌ 反模式：对大集合做 $lookup，且没有索引
db.orders.aggregate([
  { $lookup: {
      from: "products",
      localField: "productId",
      foreignField: "_id",
      as: "productDetails"
  }}
])

// 如果 orders 有 1000 万条，products 有 100 万条，
// 且 productId 上没有索引，这个查询可能跑几分钟甚至超时

// ✅ 正确做法：先过滤 + 索引
// 1. 确保关联字段有索引
db.products.createIndex({ _id: 1 })

// 2. 先用 $match 减少 orders 的数据量
db.orders.aggregate([
  { $match: { createdAt: { $gte: ISODate("2024-01-01") } } },  // 先过滤
  { $limit: 1000 },  // 限制数量
  { $lookup: {
      from: "products",
      localField: "productId",
      foreignField: "_id",
      as: "productDetails"
  }}
])

⚠️

$lookup 的"假 JOIN"问题：MongoDB 的 $lookup 每次都在内存中执行子查询，不像关系型数据库有查询优化器来选择 JOIN 顺序。在 $lookup 的 from 集合很大的情况下，$lookup 会成为性能瓶颈。如果需要频繁 JOIN 的场景，应该考虑是否应该用嵌入式文档替代。

2.3 `$lookup` 的高级用法

// 管道式子查询（MongoDB 3.6+，在子查询中再加 $match 过滤）
db.orders.aggregate([
  {
    $lookup: {
      from: "products",
      let: { order_productId: "$productId" },  // 定义局部变量
      pipeline: [
        { $match: { $expr: { $eq: ["$_id", "$$order_productId"] } } },
        { $project: { name: 1, price: 1, _id: 0 } }  // 只返回需要的字段
      ],
      as: "productDetails"
    }
  }
])

// 多字段关联
{
  $lookup: {
    from: "inventory",
    let: { p: "$productId", w: "$warehouseId" },
    pipeline: [
      { $match: {
          $expr: {
            $and: [
              { $eq: ["$productId", "$$p"] },
              { $eq: ["$warehouseId", "$$w"] }
            ]
          }
        }
      }
    ],
    as: "inventoryInfo"
  }
}

三、管道优化与性能 🟡

3.1 阶段顺序优化原则

MongoDB 的管道优化器会重排某些阶段，但了解这些原则可以写出更高效的查询：

// ✅ 优化前：$match 在 $sort 后面
db.sales.aggregate([
  { $sort: { saleDate: -1 } },         // 先排序所有数据
  { $match: { status: "completed" } }, // 再过滤（浪费了大量排序工作）
  { $limit: 100 }
])

// ✅ 优化后：$match 靠前
db.sales.aggregate([
  { $match: { status: "completed" } },  // 先过滤，减少数据量
  { $sort: { saleDate: -1 } },          // 再排序
  { $limit: 100 }
])

// MongoDB 会自动做类似优化（称为 Coalescence）：
// $match + $sort → $sort 在前，但 $match 会被下推到 sort 之前

💡

一个实用的原则：早过滤、早排序、早限制。任何能减少数据量的阶段（$match、$limit、$unwind）都应该尽量靠前。

3.2 内存限制与 `allowDiskUse`

// 默认内存限制：100MB
// 如果聚合超过限制，会报错：
// "Exceeded memory limit of 100MB. Allow disk use to avoid this error."

// 解决方案：允许使用磁盘
db.largeCollection.aggregate([
  { $group: { _id: "$category", count: { $sum: 1 } } },
  { $sort: { count: -1 } }
], { allowDiskUse: true })

// ⚠️ 但注意：allowDiskUse 开启后，排序操作变成外部排序（磁盘），
// 性能会大幅下降。如果数据量大且经常超内存，应该优化管道设计。

3.3 索引覆盖聚合查询

// 索引覆盖的聚合：只在索引中完成所有计算，不需要访问原始文档
db.products.createIndex({ category: 1, price: 1, name: 1 })

db.products.aggregate([
  { $match: { category: "electronics" } },  // 使用索引
  {
    $group: {
      _id: "$category",
      avgPrice: { $avg: "$price" }  // 只在索引中计算，不回表
    }
  }
])

四、`$facet` 与多维度聚合 🟢

4.1 `$facet` 的使用场景

$facet 允许在同一阶段对数据进行多维度聚合，一次查询返回多个独立的聚合结果：

// 场景：电商仪表盘，需要同时获取多个统计维度
db.orders.aggregate([
  { $match: { createdAt: { $gte: ISODate("2024-01-01") } } },
  {
    $facet: {
      // 维度一：按月份统计销售额
      monthlySales: [
        { $group: {
            _id: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
            total: { $sum: "$amount" },
            count: { $sum: 1 }
        }},
        { $sort: { _id: 1 } }
      ],
      // 维度二：TOP 10 商品
      topProducts: [
        { $unwind: "$items" },
        { $group: { _id: "$items.productId", totalSold: { $sum: "$items.quantity" } } },
        { $sort: { totalSold: -1 } },
        { $limit: 10 }
      ],
      // 维度三：用户等级分布
      userLevelDistribution: [
        { $group: { _id: "$userLevel", count: { $sum: 1 } } },
        { $sort: { count: -1 } }
      ],
      // 维度四：平均客单价
      averageMetrics: [
        {
          $group: {
            _id: null,
            avgOrderValue: { $avg: "$amount" },
            maxOrderValue: { $max: "$amount" },
            minOrderValue: { $min: "$amount" }
          }
        }
      ]
    }
  }
])

// 输出：一个文档包含所有四个维度的结果

💡

$facet 的每个子管道是完全独立的，输入相同的原始数据。如果子管道之间有重复的计算（如多个子管道都需要 $match），可以考虑在 $facet 之前先 $match 一次，减少数据量。

4.2 `$bucket` 和 `$bucketAuto`

// $bucket：按指定边界分组
db.products.aggregate([
  {
    $bucket: {
      groupBy: "$price",
      boundaries: [0, 100, 500, 1000, 5000],
      default: "其他",  // 不在 boundaries 内的值归入此组
      output: {
        count: { $sum: 1 },
        products: { $push: "$name" }
      }
    }
  }
])

// $bucketAuto：自动均分
db.products.aggregate([
  {
    $bucketAuto: {
      groupBy: "$price",
      buckets: 5,  // 自动分为 5 个 bucket，每个 bucket 的价格区间大致相等
      output: {
        count: { $sum: 1 },
        avgPrice: { $avg: "$price" }
      }
    }
  }
])

五、生产避坑

5.1 场景：聚合管道 OOM 导致服务崩溃

我们曾上线一个报表系统，用聚合管道统计用户行为数据。某天 DBA 告警：MongoDB 服务 CPU 飙升到 100%，紧接着服务不可用。

// 问题查询（开发同学写的）
db.userActions.aggregate([
  { $unwind: "$eventList" },  // 每个用户有几百到几千个事件
  { $group: { _id: "$eventType", users: { $addToSet: "$userId" } } }
])
// 假设 10 万用户，每个用户平均 1000 个事件
// unwind 后变成 1 亿条文档
// $group 的 $addToSet 需要把所有 userId 加载到内存
// 结果：OOM

// 正确的做法：先用 $match 过滤时间范围和事件类型
db.userActions.aggregate([
  { $match: { eventDate: { $gte: ISODate("2024-01-01") } } },  // 先过滤
  { $unwind: "$eventList" },
  { $match: { "eventList.type": { $in: ["click", "view"] } } },  // 再过滤事件类型
  { $group: { _id: "$eventList.type", count: { $sum: 1 } } }
], { allowDiskUse: true, maxTimeMS: 30000 })  // 加超时限制

5.2 避坑清单

陷阱	风险	应对策略
`$lookup` 关联大表无索引	OOM、查询超时	先 `$match` 过滤 + 确保关联字段有索引
`$unwind` 大数组	数据量爆炸	先过滤、再 unwind
聚合结果超过 100MB	OOM 报错	开启 `allowDiskUse` 或优化管道
`$sort` 在 `$limit` 之前	排序全量数据	确保 `$limit` 靠前
`$facet` 子管道重复计算	性能浪费	把共同的 `$match` 移到 `$facet` 之前
缺少 `maxTimeMS`	慢查询拖死服务	任何聚合管道都要加超时限制

【面试官心理】聚合管道是 MongoDB 的"瑞士军刀"，用得好可以替代大量的应用层处理逻辑，用不好就是性能杀手。面试中问聚合管道，我主要看两点：第一，候选人是否理解"流式处理"的模型——每个阶段的输出是下一阶段的输入；第二，候选人有没有意识到管道的边界和限制——内存限制、$lookup 的假 JOIN 本质、$unwind 的数据膨胀。能说清楚这些的，基本都有生产踩坑经验。

聚合框架（Aggregation Pipeline）#

#一、聚合管道基础 🔴

#1.1 什么是聚合管道

#1.2 核心阶段详解

#1.3 ❌ 错误示范

#1.4 标准回答

#二、$lookup 与表关联 🟡

#2.1 $lookup 的基本用法

#2.2 $lookup 的性能陷阱

#2.3 $lookup 的高级用法

#三、管道优化与性能 🟡

#3.1 阶段顺序优化原则

#3.2 内存限制与 allowDiskUse

#3.3 索引覆盖聚合查询

#四、$facet 与多维度聚合 🟢

#4.1 $facet 的使用场景

#4.2 $bucket 和 $bucketAuto

#五、生产避坑

#5.1 场景：聚合管道 OOM 导致服务崩溃

#5.2 避坑清单