当前位置：首页 > news >正文

【MongoDB篇】MongoDB的聚合框架！

news 来源：原创 2025/5/7 7:23:00

在这里插入图片描述

- 引言
- 第一节：什么是聚合框架？ 🤔
- 第二节：管道的“发动机”们——常用聚合阶段详解！⚙️
- 第三节：聚合表达式——管道中的“计算器”和“转换器” 🧮✏️
- 第四节：性能优化与考量——让聚合管道跑得更快！🏃‍♀️💨
- 第五节：总结聚合操作！🎉📊

🌟我的其他文章也讲解的比较有趣😁，如果喜欢博主的讲解方式，可以多多支持一下，感谢🤗！

其他优质专栏：【🎇SpringBoot】【🎉多线程】【🎨Redis】【✨设计模式专栏（已完结）】…等

如果喜欢作者的讲解方式，可以点赞收藏加关注，你的支持就是我的动力
✨更多文章请看个人主页：码熔burning

看之前可以先了解一下MongoDB是什么：【MongoDB篇】万字带你初识MongoDB！

引言

各位大佬们，好久不见！✨🧙‍♀️

通过之前的文章我们已经掌握了对单个文档的 CRUD 操作，这就像是你会使用基础的工具来处理一份文件。但是，如果你想对大量文档进行复杂的处理，比如统计某个年龄段的用户数量、计算商品的平均价格、找出每个城市有多少男性用户等等，简单的 find() 查询就显得力不从心了！

这时候，就轮到 MongoDB 的“瑞士军刀”——聚合框架 (Aggregation Framework) 出场了！它就像一个数据处理的工厂流水线，可以对你的数据进行一系列复杂的转换、筛选、分组和计算，最终得出你想要的“成品”！🏭➡️📦📊

第一节：什么是聚合框架？ 🤔

聚合框架是 MongoDB 提供的一个强大的、灵活的数据处理工具。它基于数据处理管道 (Data Processing Pipeline) 的概念。

你可以把聚合管道想象成一个由多个阶段 (Stage) 串联组成的流水线。集合中的文档作为“原料”，从管道的入口进入，经过第一个阶段的处理，输出的结果再作为第二个阶段的输入，依此类推，直到通过最后一个阶段，最终得到处理后的结果！

管道阶段 (Pipeline Stage) 是聚合框架的核心。每个阶段都执行特定的数据处理任务，比如：

筛选文档 (过滤掉不符合条件的) 🔍
重塑文档结构 (添加/删除/重命名字段，计算新字段) 🔨
对文档进行分组 (按某个字段分组) 🧑‍🤝‍🧑
对分组进行计算 (求和、平均值、计数等) 📊
对结果进行排序、限制数量 ⬆️⬇️✂️
展开数组 💥
关联其他集合 (类似 JOIN) 🤝

通过将这些阶段按照特定的顺序组合起来，你可以实现各种复杂的数据处理和分析任务！👍

核心方法：aggregate()

使用聚合框架，主要通过 db.collection.aggregate([pipeline], options) 方法。

[pipeline]: 一个数组，数组中的每个元素就是一个管道阶段。阶段按照数组的顺序依次执行。这是聚合管道的“蓝图”！ blueprints
options: 可选参数，用于配置聚合的行为，比如：
- allowDiskUse: <boolean>：如果设置为 true，当聚合操作内存不足时，允许将数据临时写入磁盘。处理大量数据时非常有用，但可能降低性能。默认通常为 false。💾
- cursor: <document>：指定返回结果游标的初始配置。
- readConcern: <string>：指定读取数据的隔离级别。
- writeConcern: <document>：如果聚合管道的最后一个阶段是 $out 或 $merge，可以指定写入安全级别。

use mydatabase;// 一个简单的聚合管道示例：计算所有用户的平均年龄
db.users.aggregate([{$group: { // 第一个阶段: 分组_id: null, // 按 null 分组，表示对所有文档进行一次分组avgAge: { $avg: "$age" } // 计算 age 字段的平均值}}
]);

上面的例子虽然简单，但它展示了聚合管道的基本结构：一个包含阶段的数组。

第二节：管道的“发动机”们——常用聚合阶段详解！⚙️

理解并熟练使用各种管道阶段，是掌握聚合框架的关键！哥来详细讲解一些最常用、最重要的阶段！

假设我们有一个 orders 集合，文档结构大致如下：

{"_id": ObjectId(...),"order_id": "A123","customer_id": "C001","order_date": ISODate("2023-01-15T10:00:00Z"),"items": [{ "product_id": "P001", "name": "Laptop", "price": 800, "quantity": 1 },{ "product_id": "P002", "name": "Mouse", "price": 20, "quantity": 2 }],"total_amount": 840,"status": "completed"
}

以及一个 products 集合：

{"_id": ObjectId(...),"product_id": "P001","name": "Laptop","category": "Electronics","price": 800
}

我们将使用这些集合来举例说明不同阶段的功能。

$match - 筛选文档 (过滤) 🔍📄

作用：根据指定的条件过滤输入文档，只将符合条件的文档传递给管道的下一个阶段。
位置：通常放在管道的开头，因为它可以减少后续阶段需要处理的文档数量，大大提高效率！利用索引！🚀
语法：{ $match: { <query document> } }，query document 的语法跟 find() 方法的查询条件一样！

// 找出所有状态为 "completed" 的订单
db.orders.aggregate([{ $match: { status: "completed" } }
]);// 找出所有总金额大于 1000 的订单
db.orders.aggregate([{ $match: { total_amount: { $gt: 1000 } } }
]);// 组合条件：找出状态为 "completed" 且总金额大于 500 的订单
db.orders.aggregate([{ $match: { status: "completed", total_amount: { $gt: 500 } } }
]);

$project - 重塑文档 (选择、计算字段) 🔨📄➡️📄

作用：重塑每个文档的结构，可以包含、排除、重命名字段，添加新字段（基于现有字段进行计算），或创建嵌套文档。
语法：{ $project: { <specification document> } }，规范文档使用 1 或 0 控制字段包含/排除，或者使用表达式定义新字段。

// 只保留 order_id 和 total_amount 字段 (并排除 _id，因为 _id 默认包含)
db.orders.aggregate([{$project: {_id: 0, // 排除 _id 字段orderId: "$order_id", // 重命名 order_id 为 orderIdamount: "$total_amount"}}
]);// 添加一个新字段：订单年限
db.orders.aggregate([{$project: {order_id: 1,orderYear: { $year: "$order_date" } // 使用日期表达式 $year 提取年份}}
]);// 计算每个订单的商品总数量
db.orders.aggregate([{$project: {order_id: 1,totalItems: { $sum: "$items.quantity" } // 使用数组表达式 $sum 计算 items 数组中所有 quantity 的总和}}
]);

$project 是进行数据转换和结构调整的神器！你可以使用各种聚合表达式在 $project 中进行计算和操作。

$group - 分组与聚合计算 🧑‍🤝‍🧑📊

作用：根据指定的分组键 (_id) 对文档进行分组，然后对每个分组应用一个或多个累加器表达式 (Accumulator Expressions) 来计算聚合值。
语法：{ $group: { _id: <expression>, <field1>: { <accumulator1> }, ... } }
- _id: 分组键。可以是一个字段名 ("$field"), 多个字段的组合文档 ({ field1: "$field1", ... }), 或 null (对所有文档进行一次分组)。
- <field>: { <accumulator> }：定义分组后新增的字段及其计算方式。<accumulator> 是累加器表达式。
常用累加器表达式：
- $sum: 计算总和。可以是数值字段，或 1 来计数。
- $avg: 计算平均值。
- $min: 找到最小值。
- $max: 找到最大值。
- $count: 计数分组中的文档数量 (MongoDB 3.4+)。
- $first: 获取分组中的第一个文档的某个字段值 (取决于分组前的顺序)。
- $last: 获取分组中的最后一个文档的某个字段值 (取决于分组前的顺序)。
- $push: 将分组中文档的某个字段值添加到结果文档的一个数组中。
- $addToSet: 将分组中文档的某个字段值添加到结果文档的一个数组中，并确保唯一性。
```
// 统计每个客户的订单数量
db.orders.aggregate([{$group: {_id: "$customer_id", // 按 customer_id 分组orderCount: { $sum: 1 } // 每个文档加 1，然后求和，就是订单数量}}
]);// 统计每个客户的总消费金额
db.orders.aggregate([{$group: {_id: "$customer_id",totalSpend: { $sum: "$total_amount" } // 对每个分组的 total_amount 求和}}
]);// 统计不同状态的订单数量和总金额
db.orders.aggregate([{$group: {_id: "$status", // 按 status 分组count: { $sum: 1 },total: { $sum: "$total_amount" }}}
]);// 统计每个订单的商品名称列表 (使用 $push)
// 注意：这里需要先展开 items 数组才能访问每个 item 的 name
db.orders.aggregate([{ $unwind: "$items" }, // 先展开 items 数组{$group: {_id: "$order_id", // 按 order_id 分组productNames: { $push: "$items.name" } // 将每个 item 的 name 收集到一个数组中}}
]);
```
$group 是聚合框架中最常用的阶段之一，用于实现各种统计和分组分析！

$sort - 排序 ⬆️⬇️

作用：对管道中的文档进行排序，就像 find().sort() 一样。
位置：放在 $group 之后通常用于对分组结果排序。放在 $limit 或 $skip 之前通常用于配合分页。
语法：{ $sort: { <field1>: 1/-1, ... } }

// 找出订单金额最高的 5 个客户 (先按总金额降序，然后限制 5 个)
db.orders.aggregate([{$group: {_id: "$customer_id",totalSpend: { $sum: "$total_amount" }}},{ $sort: { totalSpend: -1 } }, // 按总金额降序排序{ $limit: 5 } // 限制前 5 个
]);

$limit - 限制数量 ✂️

作用：限制通过管道的文档数量。
位置：通常放在 $sort 之后用于分页或获取 Top N。
语法：{ $limit: <number> }

// 获取最近的 10 个订单 (假设文档按插入顺序排列或者前面有 $sort)
db.orders.aggregate([{ $sort: { order_date: -1 } }, // 按日期降序{ $limit: 10 } // 限制前 10 个
]);

$skip - 跳过 🏃‍♂️

作用：跳过指定数量的文档。
位置：通常与 $sort 和 $limit 结合用于分页。
语法：{ $skip: <number> }

// 分页：获取订单的第二页，每页 10 条 (假设已排序)
db.orders.aggregate([{ $sort: { order_date: -1 } },{ $skip: 10 }, // 跳过第一页 (10 条){ $limit: 10 } // 获取第二页 (10 条)
]);

$unwind - 展开数组 💥

作用：将输入文档中的一个数组字段“展开”。对于输入文档中的数组字段，$unwind 阶段会为数组中的每个元素输出一个新文档。新文档与原文档相同，只是数组字段的值被替换为数组中的一个元素。
用处：当你需要对数组中的每个元素独立进行处理或分析时，$unwind 是 필수 (必须的)！
语法：{ $unwind: "$<array field name>" }

// 展开 orders 集合的 items 数组
db.orders.aggregate([{ $unwind: "$items" } // 将每个 order 的 items 数组展开// 展开后，一个有 2 个 items 的订单会变成 2 个文档，每个文档的 items 字段分别是数组中的一个元素
]);// 展开 items 数组后，计算每种产品的总销售数量
db.orders.aggregate([{ $unwind: "$items" }, // 展开 items 数组{$group: { // 按 products_id 分组_id: "$items.product_id",totalSold: { $sum: "$items.quantity" } // 计算每个产品的总销售数量}}
]);

$unwind 是处理包含数组字段的文档的利器！

$lookup - 关联查询 (Left Outer Join) 🤝

作用：对集合执行左外连接 (Left Outer Join)。它允许你将来自另一个集合的文档合并到当前管道中的文档中，实现类似关系型数据库 JOIN 的功能！
语法：

{$lookup: {from: "<collection to join>",       // 要连接的另一个集合localField: "<field from input documents>", // 当前集合中用于连接的字段foreignField: "<field from the documents of the 'from' collection>", // 'from' 集合中用于连接的字段as: "<output array field name>"   // 输出的新数组字段名，匹配到的文档会放入这个数组}
}

// 将订单文档与产品文档进行关联，获取每个订单中产品的详细信息
db.orders.aggregate([{ $unwind: "$items" }, // 先展开 items 数组，以便按 product_id 关联{$lookup: {from: "products", // 连接 products 集合localField: "items.product_id", // orders (当前) 集合中用于连接的字段foreignField: "product_id", // products (from) 集合中用于连接的字段as: "productInfo" // 将匹配到的产品信息放入 productInfo 数组}},// $lookup 返回的是一个数组，通常需要 $unwind 展开，如果确定只有一个匹配项的话{ $unwind: "$productInfo" },{$project: { // 重塑输出文档结构_id: 0,order_id: 1,item_name: "$items.name",item_quantity: "$items.quantity",product_category: "$productInfo.category", // 从关联到的 productInfo 中获取 categoryproduct_price: "$productInfo.price" // 从关联到的 productInfo 中获取 price}}
]);

$lookup 是实现集合之间关联查询的关键阶段！

$out - 输出到新集合 📤📁

作用：将聚合管道的最终结果写入一个新的集合。如果目标集合已存在，会覆盖整个集合！谨慎使用！ 💣
位置：必须是管道的最后一个阶段。
语法：{ $out: "<output collection name>" }
```
// 将每个客户的总消费金额计算出来，并写入一个新集合 customer_total_spend
db.orders.aggregate([{$group: {_id: "$customer_id",totalSpend: { $sum: "$total_amount" }}},{ $out: "customer_total_spend" } // 将结果输出到 customer_total_spend 集合
]);
```
$out 适合用于创建“物化视图”，比如定期生成报表数据供查询使用。

$merge - 合并到现有集合 ➡️📁

作用：将聚合管道的最终结果合并到现有的集合。比 $out 更灵活，可以指定如何处理匹配到的文档。
位置：必须是管道的最后一个阶段。
语法：

{$merge: {into: "<collection to merge into>", // 要合并到的目标集合on: "<field(s)>",                 // 用于匹配文档的字段或字段数组whenMatched: "<action>",         // 当找到匹配文档时采取的行动 ('replace', 'merge', 'fail', 'pipeline')whenNotMatched: "<action>"       // 当没有找到匹配文档时采取的行动 ('insert', 'discard', 'fail')}
}

// 将计算出的客户总消费金额合并到现有的 customers 集合中
db.orders.aggregate([{$group: {_id: "$customer_id",totalSpend: { $sum: "$total_amount" }}},{$merge: {into: "customers", // 合并到 customers 集合on: "_id",         // 以 _id 字段进行匹配 (假设 customers 的 _id 是 customer_id)whenMatched: "merge", // 找到匹配项时，合并字段whenNotMatched: "insert" // 没有找到匹配项时，插入新文档}}
]);

$merge 在需要更新或同步数据时非常有用！

第三节：聚合表达式——管道中的“计算器”和“转换器” 🧮✏️

很多聚合阶段（比如 $project, $group, $match 的 $expr）都需要使用聚合表达式 (Aggregation Expressions) 来进行计算、转换或引用字段值。表达式非常强大，支持各种运算符和函数。

引用字段：使用 "$fieldName" 语法来引用当前文档中的字段值。
字面量：直接使用数值、字符串、布尔值、日期等常量。
系统变量：使用 "$ 开头的系统变量，比如 $$ROOT (当前文档的根), $$CURRENT (当前处理的字段值)。

常见的表达式类别：

算术表达式：$add, $subtract, $multiply, $divide, $mod 等。
字符串表达式：$concat, $substr, $toUpper, $toLower, $split 等。
日期表达式：$year, $month, $dayOfMonth, $hour, $minute, $second, $dayOfWeek, $dateToString 等。
逻辑表达式：$and, $or, $not, $eq, $ne, $gt, $lt, $cond (条件判断) 等。
数组表达式：$size (数组长度), $arrayElemAt (获取数组元素), $filter (过滤数组元素), $map (转换数组元素) 等。
集合表达式：$setEquals, $setIntersection, $setUnion, $setDifference, $setIsSubset 等 (用于比较数组集合)。
条件表达式：$cond (if-then-else), $ifNull (如果为 null 则使用默认值), $switch (多条件分支)。

表达式示例：

db.orders.aggregate([{$project: {_id: 0,order_id: 1,// 计算折扣价格 (假设 total_amount 是原价)discountedPrice: { $multiply: ["$total_amount", 0.9] },// 判断订单是否是今年创建的isThisYear: { $eq: [{ $year: "$order_date" }, { $year: new Date() }] },// 使用 $cond 根据状态显示不同消息message: {$cond: {if: { $eq: ["$status", "completed"] },then: "订单已完成",else: "订单处理中"}},// 将 items 数组的每个元素的 quantity 乘以 10updatedItems: {$map: {input: "$items",as: "item",in: {product_id: "$$item.product_id",name: "$$item.name",price: "$$item.price",quantity: { $multiply: ["$$item.quantity", 10] }}}}}}
]);

聚合表达式非常丰富，掌握它们能够让你在聚合管道中进行各种复杂的数据计算和转换！去查阅官方文档，探索更多强大的表达式吧！💪

第四节：性能优化与考量——让聚合管道跑得更快！🏃‍♀️💨

聚合框架功能强大，但也可能非常消耗资源。优化聚合管道的性能至关重要！

$match 阶段前移：将 $match 阶段放在管道的越前面越好！这样可以尽早减少进入后续阶段的文档数量，显著提高效率。
利用索引：聚合管道最开始的 $match 阶段和紧随其后的 $sort 阶段可以利用索引来提高性能。确保你为这些阶段使用的字段创建了合适的索引。但是，管道后面的阶段通常无法利用索引，它们操作的是内存中的文档。
$project 只保留必要的字段：在 $project 阶段，只包含你后续需要的字段。减少文档的体积可以提高管道的处理速度，特别是当数据需要溢出到磁盘时。
避免在 $sort 之前使用会改变文档结构或数量的阶段：比如在 $sort 之前使用 $project 重命名了排序字段，或者使用了 $unwind 改变了文档数量，都会影响 $sort 阶段能否利用索引。
内存限制：如前所述，聚合阶段有内存限制。对于大型聚合，使用 allowDiskUse: true 选项，但要意识到这会影响性能。优化管道以减少内存使用是更好的方法。
使用 .explain() 分析聚合管道：这是分析聚合性能的最重要工具！在你的 aggregate() 调用后面加上 .explain()，查看每个阶段的执行情况、是否使用了索引、处理了多少文档等等。通过分析执行计划，找出性能瓶颈并进行优化！📊🔬
考虑创建物化视图：对于需要频繁查询的复杂聚合结果，可以考虑使用 $out 或 $merge 将结果写入一个单独的集合（物化视图），然后直接查询这个物化视图，而不是每次都重新执行复杂的聚合管道。

第五节：总结聚合操作！🎉📊

太棒了！我们一起探索了 MongoDB 强大的聚合框架！我们了解了它作为数据处理流水线的概念，学习了 aggregate() 方法以及 $match, $project, $group, $sort, $limit, $skip, $unwind, $lookup, $out, $merge 等核心管道阶段的功能和用法！我们也了解了聚合表达式以及如何进行聚合性能优化。

核心要点回顾：

聚合框架是用于复杂数据处理和分析的管道。
管道由多个阶段组成，按顺序处理文档。
aggregate() 方法用于执行聚合管道。
常用阶段包括筛选 ($match), 重塑 ($project), 分组 ($group), 排序 ($sort), 限制 ($limit), 跳过 ($skip), 展开数组 ($unwind), 关联 ($lookup), 输出 ($out), 合并 ($merge)。
聚合表达式用于阶段内的计算和转换。
性能优化关注阶段顺序、索引利用、内存使用、使用 .explain()。