MongoDB进阶——聚合操作

概览

聚合操作

• 单一用途的聚合方法
• Map Reduce
• 聚合管道 db.collection.aggregate()

聚合表达式

• 用来操作输入文档的“公式”
• 经聚合表达式计算出的值可以被赋予输出文档中的字段
• 字段路径、系统变量、文本、表达式对象、操作符

聚合阶段

• 聚合阶段有顺序地排列在聚合管道中
• 绝大多数聚合阶段可以反复出现（out、geoNear除外）
• 处理范围可以是数据库层面和集合层面

聚合操作符

• 用来构建聚合表达式
• 通用语法：{ <operator>: [ <argument1>, <argument2> ... ] }

实战

聚合操作

语法：

db.<collection>.aggregate(<pipeline>, <options>)

• pipeline：定义了操作中使用的聚合管道阶段和聚合操作符
• options：声明了一些聚合操作的参数

聚合表达式

1. 字段路径表达式
   • $<field>：使用 $ 表示字段路径
   • $<field>.<sub-field>：使用 $ 和 . 表示内嵌文档字段
2. 系统变量表达式
   • $$<variable>：使用 $$ 来表示系统变量
   • $$CURRENT：表示管道中当前操作的文档，比如说：$$CURRENT.<field>和$<field>是等效的
3. 常量表达式
   • $literal: <value>：表示常量value

聚合管道阶段

• $project：对输入文档进行再次投影
• $match：对输入文档进行筛选
• $limit：筛选出管道内前N篇文档
• $skip：跳过管道内前N篇文档
• $unwind：展开输入文档中的数组字段
• $sort：对输入文档进行排序
• $lookup：对输入文档进行查询操作
• $group：对输入文档进行分组
• $out：将管道中的文档输出

聚合管道演示

$project：对输入文档进行再次投影

db.accounts.aggregate([
	{
		$project: {
			_id: 0, // 不返回主键_id
			balance: 1, // 返回金额
			clientName: "$name.firstName" // 返回名称中的firstName，把他映射到clientName字段
		}
	}
])

• 如果映射了一个文档中不存在的字段，则会返回null。
• $project是一个很常用的聚合阶段，可以用来灵活的控制输出文档的格式，也可以用来剔除不相关的字段，以优化聚合管道操作的性能。

$match：对输入文档进行筛选

db.accounts.aggregate([
	{
		$match: { // 筛选出name.firstName为alice的文档
			"name.firstName": "alice"
		}
	}
])

• $match中使用的文档筛选语法，和读取文档时的筛选语法相同。
• $match也是一个很常用的聚合阶段，应该尽量在聚合管道的开始阶段应用$match，这样可以减少后续阶段中需要处理的文档数量，优化聚合操作的性能。

$limit 和 skip筛选出管道内前N篇文档，跳过管道内前N篇文档

db.accounts.aggregate([
	{
		$limit: 1 // 只返回一个文档
	},
	{
		$skip: 2 // 跳过两个文档
	}
])

$unwind：展开输入文档中的数组字段

db.accounts.aggregate([
	{
		$unwind: {
			path: "$currency", // 在原文档中，currency是一个数组，$unwind会把数组展开，比如说currency包含两个值，
							   // 聚合操作之后就会打印出两个文档，两个文档中currency字段分别保存一个，其他字段相同
			includeArrayIndex: "ccyIndex" // 增加一个ccyIndex字段，值是原文档数组的下标
		}
	}
])

$sort：对输入文档进行排序

db.accounts.aggregate([
	{
		$sort: {
			balance: 1, // 正序
			"name.lastName": -1 // 倒序
		}
	}
])

$lookup：对输入文档进行查询操作

第一种语法：

$lookup: {
	from: <collection to join>,
	localField: <field from the input documents>,
	foreignField: <field from the documents of the "from" collection>,
	as: <output array field>
}

• from：同一个数据库中的另一个查询集合
• localField：管道文档中用来进行查询的字段
• foreignField：查询集合中的查询字段
• as：写入管道文档中的查询结果数组字段

其实这个很像关系型数据库的两个表之间的内连接：

db.accounts.aggregate([
	{
		$lookup: {
			from: "forex", // 相当于从表，主表是accounts
			localField: "currency", // 主表accounts的字段currency和下面从表forex的字段ccy进行相等匹配
			foreignField: "ccy",
			as: "forexData" // 把匹配结果保存到forexData字段，并保存到主表accounts
		}
	}
])

第二种语法：

$lookup: {
	from: <collection to join>,
	let: { <var_1>: <expression>, ..., <var_n>: <expression> },
	pipeline: [ <pipeline to execute on the collection to join> ],
	as: <output array field>
}

• pipeline：对查询集合中的文档使用聚合阶段进行处理
• let：对查询集合中的文档使用聚合阶段进行处理时，如果需要参考管道文档中的字段，则必须使用let参数对字段进行声明（可选参数）

db.accounts.aggregate([
	{
		$lookup: {
			from: "forex",
			pipeline: [
				{ $match: 
					{
						data: new Date("2019-12-14")
					} 
				}
			], // 把forex文档中，日期data为2019-12-14的文档，放在forexData字段，并且写入accounts集合中
			as: "forexData"
		}
	}
])

这个例子pipeline中，$match中这个匹配只针对查询集合forex，和accounts集合无关，这种称为不相关查询。

来看一下相关查询的例子：

db.accounts.aggregate([
	{
		$lookup: {
			from: "forex",
			let: { bal: "$balance" } // bal声明的是accounts集合文档中的字段
			pipeline: [
				{ $match: 
					{ $expr: 
						{ $and: 
							[
								{ $eq: [ "$date", new Date("2019-12-14") ] },
								{ $gt: [ "$$bal", 100 ] } // 对应accounts集合文档中balance字段
							]
						 }
					} 
				}
			],
			as: "forexData"
		}
	}
])

$group：对输入文档进行分组

语法：

$group: {
	_id: <expression>,
	<field1>: { <accumulator1> : <expression1> },
	...
}

• _id：定义分组规则
• field1：可以使用聚合操作符来定义新字段（可选参数）

db.transactions.aggregate([
	{
		$group: {
			_id: "$currency" // 按照currency分组查询
		}
	}
])

在不使用聚合操作符的情况下，$group可以返回管道文档中某一字段的所有（不重复的）值。

db.transactions.aggregate([
	{
		$group: {
			_id: "$currency", // 按照currency分组查询
			totalQty: { $sum: "$qty" }, // 输出文档中增加新字段totalQty，取qty字段的和
			totalNational: { $sum: { $multiply: [ "$price", "$qty" ] } }, // price * qty 的和
			avgPrice: { $avg: "$price" }, // 取price的平均数
			count: { $sum: 1 }, // 该组文档总数
			maxNotional: { $max: { $multiply: [ "$price": "$qty" ] } }, // price * qty 的最大值
			minNotional: { $max: { $multiply: [ "$price": "$qty" ] } } // price * qty 的最小值
		}
	}
])

假如说我们要对所有文档求和求平均，不进行分组，我们可以这么做：直接把 _id 设为 null

db.transactions.aggregate([
	{
		$group: {
			_id: null, // 不分组
			totalQty: { $sum: "$qty" }, // 输出文档中增加新字段totalQty，取qty字段的和
			totalNational: { $sum: { $multiply: [ "$price", "$qty" ] } }, // price * qty 的和
			avgPrice: { $avg: "$price" }, // 取price的平均数
			count: { $sum: 1 }, // 文档总数
			maxNotional: { $max: { $multiply: [ "$price": "$qty" ] } }, // price * qty 的最大值
			minNotional: { $max: { $multiply: [ "$price": "$qty" ] } } // price * qty 的最小值
		}
	}
])

$out：将管道中的文档输出

db.transactions.aggregate([
	{
		$group: {
			_id: "$symbol",
			totalNational: { $sum: { $multiply: [ "$price", "$qty" ] } }
		}
	},
	{
		$out: "output" // 把上面的查询结果写入到output集合
	}
])

如果聚合管道操作遇到错误，管道阶段不会创建新集合或是覆盖已存在的集合内容。