MongoDB聚合管道实战：$match与$group阶段的数据处理方案

MongoDB聚合管道基础与执行原理

MongoDB聚合管道（Aggregation Pipeline）是处理NoSQL数据分析和报表统计的核心功能。管道由多个阶段（Stage）组成，文档依次通过每个阶段被处理和转换，最终输出结果。这种设计类似于Unix管道，每个阶段的输出作为下一阶段的输入。

聚合管道的执行顺序直接影响性能。MongoDB对管道阶段有优化器，会自动调整某些阶段的顺序，但手动遵循”尽早过滤、尽早减少文档数量”的原则始终有效。$match放在管道最前面可以利用索引加速查询，$project减少字段后降低后续阶段的内存占用。

// 聚合管道基础语法
db.collection.aggregate([
  { $match: { status: "active" } },           // 过滤
  { $project: { name: 1, price: 1, _id: 0 } }, // 字段投影
  { $sort: { price: -1 } },                     // 排序
  { $limit: 100 },                              // 限制
  { $skip: 0 }                                  // 跳过
])

// 等价SQL: SELECT name, price FROM collection 
//          WHERE status = 'active' 
//          ORDER BY price DESC LIMIT 100

管道阶段数量建议控制在10个以内。过多的阶段增加执行计划复杂度，优化器可能无法找到最优执行路径。对于复杂的统计需求，考虑将部分逻辑前移到写入阶段，通过变更流或定时任务预计算结果。

$match阶段与索引优化

$match是管道中最常用的阶段，负责过滤文档。当$match出现在管道第一个位置时，MongoDB会尝试使用索引加速查询，与普通find查询的索引优化策略一致。

// 创建复合索引
db.orders.createIndex({ status: 1, created_at: -1 })
db.orders.createIndex({ user_id: 1, status: 1 })

// 聚合查询：某用户活跃订单统计
db.orders.aggregate([
  { $match: { 
      status: "active",
      created_at: { $gte: ISODate("2026-01-01") }
  }},
  { $group: {
      _id: "$user_id",
      orderCount: { $sum: 1 },
      totalAmount: { $sum: "$amount" },
      avgAmount: { $avg: "$amount" },
      maxAmount: { $max: "$amount" }
  }},
  { $sort: { totalAmount: -1 } },
  { $limit: 20 }
])

// explain分析执行计划
db.orders.aggregate([
  { $match: { status: "active" } },
  { $group: { _id: "$user_id", count: { $sum: 1 } } }
]).explain("executionStats")

// 关注explain输出中的关键字段：
// winningPlan.stage: "IXSCAN" 表示使用了索引
// executionStats.totalDocsExamined: 扫描文档数
// executionStats.executionTimeMillis: 执行耗时

$match中使用的时间范围查询需要确保字段类型一致。如果created_at字段在部分文档中存储为字符串、另一部分存储为Date类型，索引无法正常工作。建议在写入阶段统一日期格式，使用ISODate或BSON Date类型。

$group阶段与多维度聚合

$group是聚合管道中功能最丰富的阶段，支持$sum、$avg、$max、$min、$push、$addToSet等累加器操作符。通过多层$group嵌套可以实现多维度交叉统计。

// 1. 按日期和类目统计销售额
db.orders.aggregate([
  { $match: { status: "completed", created_at: { $gte: ISODate("2026-01-01") } } },
  { $project: {
      date: { $dateToString: { format: "%Y-%m-%d", date: "$created_at" } },
      category: "$category",
      amount: "$amount"
  }},
  { $group: {
      _id: { date: "$date", category: "$category" },
      dailyCategoryTotal: { $sum: "$amount" },
      orderCount: { $sum: 1 }
  }},
  { $group: {
      _id: "$_id.date",
      categories: { 
          $push: { 
              category: "$_id.category", 
              total: "$dailyCategoryTotal",
              count: "$orderCount"
          }
      },
      dailyGrandTotal: { $sum: "$dailyCategoryTotal" }
  }},
  { $sort: { _id: 1 } }
])

// 2. $bucket分组：按区间分桶
db.users.aggregate([
  { $bucket: {
      groupBy: "$age",
      boundaries: [0, 18, 25, 35, 50, 100],
      default: "other",
      output: {
        count: { $sum: 1 },
        avgIncome: { $avg: "$income" }
      }
  }}
])

// 3. $facet多管道并行
db.products.aggregate([
  { $facet: {
      priceStats: [
        { $group: { _id: null, avg: { $avg: "$price" }, max: { $max: "$price" } } }
      ],
      categoryStats: [
        { $group: { _id: "$category", count: { $sum: 1 } } },
        { $sort: { count: -1 } }
      ],
      brandStats: [
        { $group: { _id: "$brand", count: { $sum: 1 } } },
        { $sort: { count: -1 } },
        { $limit: 10 }
      ]
  }}
])

$facet允许在同一个聚合操作中执行多个独立的子管道，结果以对象形式返回。这在报表场景中避免了多次查询的开销，但所有子管道共享同一批输入文档，内存消耗为各子管道之和。$bucket用于将连续值划分为离散区间，适合年龄段、价格区间等分桶统计。

$lookup关联查询与$unwind展开

MongoDB作为NoSQL数据库不原生支持JOIN操作，但$lookup阶段提供了类似左外连接的能力，可以在聚合管道中关联另一个集合的文档。

// 订单关联用户信息
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $lookup: {
      from: "users",
      localField: "user_id",
      foreignField: "_id",
      as: "user_info"
  }},
  { $unwind: "$user_info" },
  { $project: {
      order_id: "$_id",
      amount: 1,
      user_name: "$user_info.name",
      user_email: "$user_info.email"
  }}
])

// 多条件关联查询
db.orders.aggregate([
  { $lookup: {
      from: "products",
      let: { product_id: "$product_id", order_date: "$created_at" },
      pipeline: [
        { $match: {
            $expr: {
              $and: [
                { $eq: ["$_id", "$$product_id"] },
                { $gte: ["$created_at", "$$order_date"] }
              ]
            }
        }},
        { $project: { name: 1, price: 1 } }
      ],
      as: "product_info"
  }}
])

// $unwind处理数组字段
db.articles.aggregate([
  { $match: { published: true } },
  { $unwind: {
      path: "$tags",
      preserveNullAndEmptyArrays: true
  }},
  { $group: {
      _id: "$tags",
      count: { $sum: 1 },
      titles: { $push: "$title" }
  }},
  { $sort: { count: -1 } }
])

$lookup的性能与关联集合的索引直接相关。foreignField字段必须建立索引，否则每次关联都会全表扫描。对于大量数据的关联查询，先在应用层分批获取ID再查询关联集合，可能比$lookup更快。preserveNullAndEmptyArrays设为true时，数组为空或null的文档不会被过滤，在统计缺失值时有用。

聚合管道性能调优与注意事项

聚合管道默认有100MB的内存限制，超过限制会报错。大数据量聚合需要允许使用磁盘空间：

// 允许写入临时磁盘文件
db.large_collection.aggregate([
  { $match: { status: "active" } },
  { $group: { _id: "$category", count: { $sum: 1 } } }
], { allowDiskUse: true })

// 分批处理大数据集
const cursor = db.orders.aggregate([
  { $match: { status: "completed" } },
  { $sort: { _id: 1 } }
], { batchSize: 1000, allowDiskUse: true })

while (cursor.hasNext()) {
  const batch = cursor.next();
  processBatch(batch);
}

// 监控聚合性能
db.runCommand({
  aggregate: "orders",
  pipeline: [
    { $match: { status: "active" } },
    { $group: { _id: "$category", total: { $sum: "$amount" } } }
  ],
  explain: true
})

allowDiskUse: true允许管道中间结果写入临时文件，代价是磁盘I/O开销增大。对于$group阶段产生的数据量超过100MB的场景，这是必要的配置。如果聚合查询是高频操作，考虑通过定时任务将结果预计算并写入物化视图集合，查询时直接读取结果集合，将O(n)的聚合降为O(1)的查找。