mongodb

04. MongoDB 聚合管道深度解析

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

Leeting Yan 2026-08-14 2 分钟阅读 921 字

聚合管道（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）

// 订单关联用户信息
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+）

// 计算移动平均和累计
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"] }
                }
            }
        }
    }
]);

// 同时生成多个统计报表
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 时间序列集合）
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 });

延伸阅读

继续阅读

探索更多技术文章

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

全部文章返回首页