mongodb

04. MongoDB 聚合管道深度解析

MongoDB 聚合管道实战: $match、$group、$lookup、$facet 等核心阶段，复杂数据分析与报表生成

Leeting Yan 8 分钟阅读 3616 字

聚合管道（Aggregation Pipeline）是 MongoDB 最强大的数据分析工具，通过将多个处理阶段串联，能够实现复杂的数据转换、分组统计和关联查询。相比 MapReduce，聚合管道更直观、性能更好。

1. 管道概念

聚合管道的核心思想是将数据处理拆分为多个阶段，每个阶段接收上一阶段输出的文档流，经过处理后传递给下一阶段。这种流式处理方式使得内存占用可控，同时也便于利用索引和并行计算。在实际生产环境中，聚合管道常用于报表生成、实时统计、数据清洗和关联分析等场景。

// 基本语法结构：阶段数组依次执行
db.collection.aggregate([
    { $stage1: { ... } },  // 阶段 1：过滤或转换
    { $stage2: { ... } },  // 阶段 2：分组或关联
    { $stage3: { ... } }   // 阶段 3：排序或输出
]);
// 数据从左到右流经每个阶段，每个阶段输出作为下一阶段输入

常见阶段速查：

阶段用途类比 SQL
$match过滤文档WHERE
$project投影/重塑SELECT
$group分组聚合GROUP BY
$sort排序ORDER BY
$limit限制条数LIMIT
$skip跳过条数OFFSET
$lookup左外连接LEFT JOIN
$unwind展开数组LATERAL JOIN
$facet多管道聚合子查询
$bucket自动分桶CASE WHEN

2. 基础聚合示例

2.1 简单分组统计

日常业务中最常见的需求是按时间维度统计订单数据。以下示例演示按月汇总订单金额、订单数量、客单价、最大最小订单金额等核心指标。

// 按月统计订单金额
db.orders.aggregate([
    { $match: { status: "completed" } },
    {
        $group: {
            _id: { $dateToString: { format: "%Y-%m", date: "$createdAt" } },
            totalRevenue: { $sum: "$total" },
            orderCount: { $sum: 1 },
            avgOrderValue: { $avg: "$total" },
            maxOrder: { $max: "$total" },
            minOrder: { $min: "$total" }
        }
    },
    { $sort: { _id: -1 } }
]);

// 输出:
// { _id: "2024-01", totalRevenue: 158000, orderCount: 320, ... }
// { _id: "2023-12", totalRevenue: 145000, orderCount: 290, ... }

2.2 嵌套字段聚合

电商系统中订单文档通常包含商品数组，统计各分类的销量和营收需要先展开数组再按分类分组。

// 统计各分类商品的总销量
db.orders.aggregate([
    { $unwind: "$items" },  // 展开订单中的商品数组
    {
        $group: {
            _id: "$items.category",
            totalQuantity: { $sum: "$items.quantity" },
            totalRevenue: { $sum: { $multiply: ["$items.price", "$items.quantity"] } }
        }
    },
    { $sort: { totalRevenue: -1 } }
]);

2.3 关联查询（$lookup）

$lookup 支持两种模式：等值匹配模式和管道模式。等值匹配适用于简单的字段对应关系，管道模式则支持更复杂的关联条件和子查询逻辑。

// 订单关联用户信息
db.orders.aggregate([
    { $match: { status: "pending" } },
    {
        $lookup: {
            from: "users",           // 关联集合
            localField: "userId",    // 订单字段
            foreignField: "_id",     // 用户字段
            as: "userInfo"           // 输出数组字段名
        }
    },
    { $unwind: "$userInfo" },       // 数组展开为对象
    {
        $project: {
            orderId: 1,
            total: 1,
            "userName": "$userInfo.name",
            "userEmail": "$userInfo.email"
        }
    }
]);

// 关联条件查询 (MongoDB 3.6+)
db.orders.aggregate([
    {
        $lookup: {
            from: "products",
            let: { productId: "$productId" },
            pipeline: [
                { $match: { $expr: { $eq: ["$_id", "$$productId"] } } },
                { $project: { name: 1, price: 1 } }
            ],
            as: "product"
        }
    }
]);

3. 高级聚合

3.1 窗口函数（MongoDB 5.0+）

窗口函数在聚合管道中引入了跨行计算能力，特别适合时间序列分析、排名计算和滑动统计。$setWindowFields 阶段支持按分区、排序和窗口范围定义计算规则。

// 计算移动平均和累计
db.sales.aggregate([
    { $sort: { date: 1 } },
    {
        $setWindowFields: {
            partitionBy: "$productId",
            sortBy: { date: 1 },
            output: {
                movingAvg7d: {
                    $avg: "$amount",
                    window: { range: [-3, 3], unit: "day" }
                },
                cumulativeSum: {
                    $sum: "$amount",
                    window: { documents: ["unbounded", "current"] }
                }
            }
        }
    }
]);

3.2 $facet 多维度聚合

$facet 允许在同一个输入上并行执行多个子聚合管道，非常适合仪表盘类的一次查询多维度统计需求。

// 同时生成多个统计报表
db.products.aggregate([
    {
        $facet: {
            byCategory: [
                { $group: { _id: "$category", count: { $sum: 1 }, avgPrice: { $avg: "$price" } } }
            ],
            byPriceRange: [
                {
                    $bucket: {
                        groupBy: "$price",
                        boundaries: [0, 50, 100, 500, 1000],
                        default: "1000+",
                        output: { count: { $sum: 1 }, avgRating: { $avg: "$rating" } }
                    }
                }
            ],
            topRated: [
                { $sort: { rating: -1 } },
                { $limit: 10 },
                { $project: { name: 1, rating: 1, price: 1 } }
            ]
        }
    }
]);

3.3 时间序列聚合

MongoDB 5.x 引入原生时间序列集合，配合 $dateTrunc 等操作符可以高效处理时序数据。

// 按时间窗口聚合（MongoDB 5.x 时间序列集合）
db.sensor_readings.aggregate([
    {
        $group: {
            _id: {
                sensor: "$sensor_id",
                hour: { $dateTrunc: { date: "$timestamp", unit: "hour" } }
            },
            avgTemp: { $avg: "$temperature" },
            maxTemp: { $max: "$temperature" },
            readings: { $count: {} }
        }
    }
]);

4. 聚合优化

聚合管道的性能高度依赖阶段的编排顺序和数据量。核心优化原则是尽早缩减数据规模，充分利用索引，减少内存占用。

优化技巧说明
$match 放前面尽早过滤，减少后续处理数据量
利用索引$match 和 $sort 放在开头可用索引
$project 精简只保留需要的字段
$limit 紧跟 $sort只排序需要的前 N 条
避免 $unwind 大数据可能导致内存溢出
allowDiskUse大数据聚合允许使用磁盘
db.orders.aggregate([...], { allowDiskUse: true });

5. Stage 深度解析

5.1 $match 索引利用

$match 是唯一能利用索引的阶段。将 $match 置于管道最前端，MongoDB 可以使用集合索引直接过滤，仅将匹配文档传递给后续阶段。如果 $match 放置在 $project 或 $group 之后，则必须扫描全部中间结果。

// 推荐：$match 前置，利用 createdAt 索引
db.orders.aggregate([
    { $match: { createdAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2024-02-01") }, status: "completed" } },
    { $group: { _id: "$category", total: { $sum: "$amount" } } }
]);

5.2 $group 内存限制与优化

$group 阶段默认使用内存上限为 100MB。当分组结果超过此限制时，不加 allowDiskUse 会报错。对于高基数分组字段，应考虑使用复合索引或预聚合策略降低内存压力。

// 允许磁盘溢出，适合超大分组
db.events.aggregate([
    { $match: { eventType: "click" } },
    { $group: { _id: "$userId", clickCount: { $sum: 1 } } }
], { allowDiskUse: true, maxTimeMS: 60000 });

5.3 $lookup 嵌套查询性能

$lookup 的管道模式功能强大，但每个输入文档都会触发一次子查询，数据量大时可能成为性能瓶颈。建议在外层 $match 先过滤输入集，同时在 $lookup 的 pipeline 中也加入 $match 缩小关联集合扫描范围。

// 优化：关联双方均先过滤
db.orders.aggregate([
    { $match: { status: "shipped" } },
    {
        $lookup: {
            from: "shipments",
            let: { oid: "$_id" },
            pipeline: [
                { $match: { $expr: { $eq: ["$orderId", "$$oid"] } } },
                { $project: { carrier: 1, trackingNo: 1, shippedAt: 1 } }
            ],
            as: "shipment"
        }
    },
    { $unwind: { path: "$shipment", preserveNullAndEmptyArrays: true } }
]);

5.4 $unwind 使用陷阱与 preserveNullAndEmptyArrays

默认 $unwind 会丢弃空数组或缺失字段的文档，导致数据丢失。生产环境中务必使用 preserveNullAndEmptyArrays: true，确保不会意外过滤数据。

// 安全用法：保留空数组文档
db.orders.aggregate([
    { $unwind: { path: "$tags", preserveNullAndEmptyArrays: true } },
    { $group: { _id: "$tags", count: { $sum: 1 } } }
]);

6. 管道优化策略

6.1 Stage 顺序原则

Stage 顺序直接影响性能和正确性。最佳实践是遵循以下优先级排列：先 $match 过滤，再 $project 精简字段，然后 $unwind 展开，接着 $lookup 关联，最后 $group 聚合和 $sort 排序。

// 优化顺序示例
db.orders.aggregate([
    { $match: { status: "completed", createdAt: { $gte: ISODate("2024-01-01") } } },
    { $project: { userId: 1, items: 1, total: 1 } },
    { $unwind: "$items" },
    { $group: { _id: "$items.sku", revenue: { $sum: { $multiply: ["$items.price", "$items.quantity"] } } } },
    { $sort: { revenue: -1 } },
    { $limit: 20 }
]);

6.2 explain(“executionStats”) 分析聚合计划

使用 explain("executionStats") 可以查看聚合管道是否命中索引、各阶段文档处理数量和执行时间，是诊断性能问题的首选工具。

db.orders.explain("executionStats").aggregate([
    { $match: { status: "completed" } },
    { $group: { _id: "$category", total: { $sum: "$amount" } } }
]);

6.3 allowDiskUse 与内存限制

允许磁盘写入虽然可以避免内存溢出，但会显著降低聚合速度。建议仅在离线报表或一次性大数据任务中启用，实时查询应通过分片策略或预聚合解决。

db.large_collection.aggregate([
    { $group: { _id: "$field", count: { $sum: 1 } } }
], {
    allowDiskUse: true,
    cursor: { batchSize: 1000 },
    maxTimeMS: 300000
});

7. 窗口函数深度应用

7.1 排序窗口与排名

窗口函数支持按文档范围或值范围定义窗口，可用于计算排名、百分比和相邻行差异分析。

// 按品类计算销售额排名和累计占比
db.sales.aggregate([
    { $sort: { category: 1, amount: -1 } },
    {
        $setWindowFields: {
            partitionBy: "$category",
            sortBy: { amount: -1 },
            output: {
                rank: { $rank: {} },
                denseRank: { $denseRank: {} },
                cumulativeRevenue: {
                    $sum: "$amount",
                    window: { documents: ["unbounded", "current"] }
                }
            }
        }
    }
]);

7.2 移动平均与累积计算

移动平均是金融和监控领域最常用的指标之一。通过调整 window 参数中的 range 或 documents，可以灵活定义滑动窗口大小。

// 七天移动平均与同比累积
db.daily_metrics.aggregate([
    { $sort: { date: 1 } },
    {
        $setWindowFields: {
            partitionBy: "$metricName",
            sortBy: { date: 1 },
            output: {
                movingAvg7d: {
                    $avg: "$value",
                    window: { documents: [-6, 0] }
                },
                movingAvg30d: {
                    $avg: "$value",
                    window: { documents: [-29, 0] }
                },
                cumulative: {
                    $sum: "$value",
                    window: { documents: ["unbounded", "current"] }
                }
            }
        }
    }
]);

8. 聚合与业务场景

8.1 电商订单统计分析

电商数据平台的核心指标包括 GMV、UV、客单价和转化率。聚合管道可以一次性计算这些指标，支持运营决策。

// 计算月度 GMV、UV、客单价和支付转化率
db.orders.aggregate([
    { $match: { createdAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2024-02-01") } } },
    {
        $facet: {
            gmv: [
                { $match: { status: { $in: ["paid", "shipped", "completed"] } } },
                { $group: { _id: null, gmv: { $sum: "$total" }, orderCount: { $sum: 1 } } }
            ],
            uv: [
                { $group: { _id: "$userId" } },
                { $count: "uv" }
            ],
            conversion: [
                { $group: { _id: "$status", count: { $sum: 1 } } }
            ]
        }
    }
]);

8.2 用户行为漏斗分析

漏斗分析是增长运营和产品优化的关键手段。通过 $facet 或自连接管道，可以统计各步骤的留存和流失情况。

// 注册 -> 浏览商品 -> 加购 -> 支付 漏斗
db.user_events.aggregate([
    { $match: { eventAt: { $gte: ISODate("2024-01-01"), $lt: ISODate("2024-02-01") } } },
    {
        $group: {
            _id: "$userId",
            registered: { $max: { $cond: [{ $eq: ["$event", "register"] }, 1, 0] } },
            viewed: { $max: { $cond: [{ $eq: ["$event", "view_product"] }, 1, 0] } },
            added: { $max: { $cond: [{ $eq: ["$event", "add_cart"] }, 1, 0] } },
            paid: { $max: { $cond: [{ $eq: ["$event", "purchase"] }, 1, 0] } }
        }
    },
    {
        $group: {
            _id: null,
            totalUsers: { $sum: 1 },
            viewed: { $sum: "$viewed" },
            added: { $sum: "$added" },
            paid: { $sum: "$paid" }
        }
    },
    {
        $project: {
            totalUsers: 1,
            viewRate: { $multiply: [{ $divide: ["$viewed", "$totalUsers"] }, 100] },
            addRate: { $multiply: [{ $divide: ["$added", "$totalUsers"] }, 100] },
            payRate: { $multiply: [{ $divide: ["$paid", "$totalUsers"] }, 100] }
        }
    }
]);

8.3 地理空间聚合

结合 $geoNear 和 $group，可以实现基于位置的商圈分析、附近热点统计等空间计算需求。

// 统计各商圈内订单数量和平均消费
db.orders.aggregate([
    {
        $geoNear: {
            near: { type: "Point", coordinates: [121.4737, 31.2304] },
            distanceField: "distance",
            maxDistance: 10000,
            spherical: true
        }
    },
    {
        $bucket: {
            groupBy: "$distance",
            boundaries: [0, 1000, 3000, 5000, 10000],
            default: "far",
            output: {
                orderCount: { $sum: 1 },
                avgAmount: { $avg: "$total" }
            }
        }
    }
]);

9. 聚合性能调优

9.1 分片集群聚合策略

在分片集群上执行聚合时，如果管道以 $match 开头且包含片键，MongoDB 可以将查询路由到特定分片，避免全分片扫描。对于不包含片键的聚合，mongos 需要从所有分片拉取数据并在本地合并，开销较大。

// 片键查询：仅命中目标分片
db.orders.aggregate([
    { $match: { region: "east", createdAt: { $gte: ISODate("2024-01-01") } } },
    { $group: { _id: "$city", total: { $sum: "$amount" } } }
]);

9.2 $merge 与 $out 结果物化

$out 将聚合结果写入新集合，覆盖原有数据。$merge 则支持按需更新、插入或合并到已有集合，更适合生产环境中的增量报表。

// 使用 $merge 将日报数据合并到汇总表
db.daily_sales.aggregate([
    { $match: { date: { $gte: ISODate("2024-01-01"), $lt: ISODate("2024-02-01") } } },
    { $group: { _id: "$category", monthlyRevenue: { $sum: "$revenue" }, orders: { $sum: "$orders" } } },
    {
        $merge: {
            into: "monthly_summary",
            on: "_id",
            whenMatched: "merge",
            whenNotMatched: "insert"
        }
    }
]);

9.3 视图与物化视图

MongoDB 视图是基于聚合管道定义的只读虚拟集合，适合封装复杂查询逻辑。对于高频访问的聚合结果，可以配合定时任务将 $merge 或 $out 结果写入物化集合，用空间换时间。

// 创建只读视图
db.createView("active_users_view", "users", [
    { $match: { lastLogin: { $gte: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) } } },
    { $project: { name: 1, email: 1, lastLogin: 1 } }
]);

10. 驱动集成实战

10.1 Python PyMongo 聚合

PyMongo 是 Python 生态中使用最广泛的 MongoDB 驱动。其 aggregate 方法返回游标对象，支持迭代器协议遍历结果。

from pymongo import MongoClient
from datetime import datetime

client = MongoClient("mongodb://localhost:27017")
db = client["shop"]

pipeline = [
    {"$match": {"status": "completed", "createdAt": {"$gte": datetime(2024, 1, 1)}}},
    {"$group": {"_id": "$category", "total": {"$sum": "$amount"}, "count": {"$sum": 1}}},
    {"$sort": {"total": -1}},
    {"$limit": 10}
]

result = list(db.orders.aggregate(pipeline))
for doc in result:
    print(f"{doc['_id']}: {doc['total']} ({doc['count']} 笔)")
client.close()

10.2 Node.js Mongoose Aggregate

Mongoose 提供了链式聚合构建器，代码可读性更强。同时支持在聚合结果上继续调用 Promise 方法。

const mongoose = require("mongoose");
await mongoose.connect("mongodb://localhost:27017/shop");

const Order = mongoose.model("Order", new mongoose.Schema({}, { strict: false }));

const result = await Order.aggregate()
    .match({ status: "completed", createdAt: { $gte: new Date("2024-01-01") } })
    .group({ _id: "$category", total: { $sum: "$amount" }, count: { $sum: 1 } })
    .sort({ total: -1 })
    .limit(10);

console.log(result);
await mongoose.disconnect();

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「mongodb」更多文章

  1. 数据生命周期、TTL 与冷热归档
  2. $graphLookup 与层次结构建模
  3. GridFS 与大文件存储实践