MongoDB聚合管道实战：数据分组统计与多阶段数据处理方案

MongoDB聚合管道（Aggregation Pipeline）是处理复杂数据查询和统计的核心工具。相比简单的find查询，聚合管道通过多个阶段（stage）串联，对文档进行过滤、转换、分组、排序等操作，实现类似SQL中GROUP BY、JOIN、HAVING的功能。本文通过实际案例演示常用聚合阶段的使用方法和组合技巧。

聚合管道基础语法

聚合管道的基本语法是db.collection.aggregate([stage1, stage2, ...])，每个stage接收上一阶段的输出文档流作为输入。文档依次经过每个阶段处理，最终输出结果集。管道阶段顺序影响性能，将$match放在前面可以减少后续阶段处理的数据量。

// 基础语法
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } },
  { $sort: { total: -1 } },
  { $limit: 10 }
])

// 等价SQL:
// SELECT customerId, SUM(amount) as total
// FROM orders
// WHERE status = 'completed'
// GROUP BY customerId
// ORDER BY total DESC
// LIMIT 10

$match与$project阶段

$match阶段过滤文档，语法与find查询的查询条件相同。将$match放在管道前面可以利用索引加速，减少后续阶段的数据量。$project阶段选择、添加或重命名字段，控制输出文档的结构。

// $match过滤 + $project字段选择
db.orders.aggregate([
  {
    $match: {
      status: "completed",
      createdAt: { $gte: ISODate("2026-01-01"), $lt: ISODate("2026-10-01") },
      amount: { $gte: 100 }
    }
  },
  {
    $project: {
      orderId: 1,
      customerId: 1,
      amount: 1,
      status: 1,
      // 计算字段
      taxAmount: { $multiply: ["$amount", 0.13] },
      totalAmount: { $add: ["$amount", { $multiply: ["$amount", 0.13] }] },
      // 格式化日期
      orderMonth: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
      // 排除字段
      _id: 0
    }
  }
])

// $match中使用聚合表达式
db.orders.aggregate([
  {
    $match: {
      $expr: { $gt: ["$amount", { $multiply: ["$basePrice", "$quantity"] }] }
    }
  }
])

$group分组统计

$group是聚合管道中最常用的阶段，按指定字段分组并对每组应用累加器操作。_id字段定义分组键，可以是单个字段或复合键。常用累加器：$sum求和、$avg平均、$max/$min极值、$push收集值到数组、$addToSet收集不重复值。

// 按客户分组统计订单
db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $group: {
      _id: "$customerId",
      orderCount: { $sum: 1 },
      totalAmount: { $sum: "$amount" },
      avgAmount: { $avg: "$amount" },
      maxAmount: { $max: "$amount" },
      minAmount: { $min: "$amount" },
      orderIds: { $push: "$orderId" },
      uniqueProducts: { $addToSet: "$productId" }
    }
  },
  { $sort: { totalAmount: -1 } }
])

// 复合分组键：按客户和月份分组
db.orders.aggregate([
  {
    $group: {
      _id: {
        customerId: "$customerId",
        month: { $dateToString: { format: "%Y-%m", date: "$createdAt" } }
      },
      monthlyTotal: { $sum: "$amount" },
      orderCount: { $sum: 1 }
    }
  },
  {
    $project: {
      _id: 0,
      customerId: "$_id.customerId",
      month: "$_id.month",
      monthlyTotal: 1,
      orderCount: 1
    }
  },
  { $sort: { customerId: 1, month: 1 } }
])

多阶段管道组合实战

实际业务中通常需要组合多个阶段完成复杂统计。以下示例实现电商订单的多维度分析：按月份统计各品类的销售额和占比。

// 订单关联产品信息后分组统计
db.orders.aggregate([
  // 1. 过滤时间范围
  {
    $match: {
      status: "completed",
      createdAt: {
        $gte: ISODate("2026-01-01"),
        $lt: ISODate("2026-10-01")
      }
    }
  },
  // 2. 展开商品数组
  { $unwind: "$items" },
  // 3. 关联产品集合
  {
    $lookup: {
      from: "products",
      localField: "items.productId",
      foreignField: "_id",
      as: "productInfo"
    }
  },
  // 4. 展开关联结果
  { $unwind: "$productInfo" },
  // 5. 添加计算字段
  {
    $addFields: {
      itemTotal: { $multiply: ["$items.price", "$items.quantity"] },
      month: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
      category: "$productInfo.category"
    }
  },
  // 6. 按月份和品类分组
  {
    $group: {
      _id: { month: "$month", category: "$category" },
      revenue: { $sum: "$itemTotal" },
      orderCount: { $sum: 1 },
      avgOrderValue: { $avg: "$itemTotal" }
    }
  },
  // 7. 按月份分组，品类收入组成数组
  {
    $group: {
      _id: "$_id.month",
      categories: {
        $push: {
          category: "$_id.category",
          revenue: "$revenue",
          orderCount: "$orderCount"
        }
      },
      totalRevenue: { $sum: "$revenue" }
    }
  },
  // 8. 计算各品类占比
  {
    $addFields: {
      categories: {
        $map: {
          input: "$categories",
          as: "cat",
          in: {
            category: "$$cat.category",
            revenue: "$$cat.revenue",
            orderCount: "$$cat.orderCount",
            percentage: {
              $round: [
                { $multiply: [
                  { $divide: ["$$cat.revenue", "$totalRevenue"] },
                  100
                ]},
                2
              ]
            }
          }
        }
      }
    }
  },
  // 9. 排序
  { $sort: { _id: 1 } }
])

聚合管道性能优化

聚合管道的性能优化有几个关键原则。第一，尽早使用$match减少数据量，管道开头有$match可以利用索引。第二，$project在$group之前剔除不需要的字段，减少内存占用。第三，$group操作如果分组基数过大，可能导致内存溢出，可通过allowDiskUse: true允许写入临时文件。

// 允许使用磁盘
db.orders.aggregate(pipeline, { allowDiskUse: true })

// 查看聚合执行计划
db.orders.aggregate([
  { $match: { status: "completed" } },
  { $group: { _id: "$customerId", total: { $sum: "$amount" } } }
], { explain: true })

// 创建聚合查询优化索引
db.orders.createIndex({ status: 1, createdAt: -1, amount: 1 })
db.orders.createIndex({ customerId: 1 })

// 使用$facet并行执行多个聚合
db.orders.aggregate([
  { $match: { status: "completed" } },
  {
    $facet: {
      "byCategory": [
        { $group: { _id: "$category", total: { $sum: "$amount" } } }
      ],
      "byMonth": [
        { $group: {
          _id: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
          total: { $sum: "$amount" }
        }}
      ],
      "overallStats": [
        { $group: {
          _id: null,
          totalRevenue: { $sum: "$amount" },
          avgOrderValue: { $avg: "$amount" },
          orderCount: { $sum: 1 }
        }}
      ]
    }
  }
])

MongoDB聚合管道在数据分析报表、实时监控统计、ETL数据清洗等场景中具有很高的实用性。相比MapReduce（MongoDB 5.0已废弃），聚合管道语法更简洁、性能更好、支持索引加速。对于超大规模数据集的离线分析，建议使用MongoDB的Atlas Trigger配合聚合管道，或将数据导出到专用分析引擎处理。