聚合操作

MongoDB的聚合操作是通过数据处理管道(pipeline)来实现的,一次操作可以通过多个管道来处理，且管道是有顺序的

聚合操作可用于实现分组, 排序, 数值运算, 条件筛选，多表关联查询等

聚合管道包含非常丰富的聚合阶段，下面是最常用的聚合阶段

$project：修改输入文档的结构。可以用来重命名、增加或删除域，也可以用于创建计算结果以及嵌套文档。
$match：用于过滤数据，只输出符合条件的文档。$match使用MongoDB的标准查询操作。
$limit：用来限制MongoDB聚合管道返回的文档数。
$skip：在聚合管道中跳过指定数量的文档，并返回余下的文档。
$unwind：将文档中的某一个数组类型字段拆分成多条，每条包含数组中的一个值。
$group：将集合中的文档分组，可用于统计结果。
$sort：将输入文档排序后输出。
$geoNear：输出接近某一地理位置的有序文档

语法格式

db.collection.aggregate([
{pipeline_1},
{pipeline_2}
….
])

1 聚合-$group

$group 聚合操作符主要作用是对文档中特定字段进行分组，搭配下面的操作符对结果进行计算

db.sales.drop()
db.sales.insertMany([
  { "_id" : 1, "item" : "abc", "price" : NumberDecimal("10"), "quantity" : NumberInt("2"), "date" : ISODate("2014-03-01T08:00:00Z") },
  { "_id" : 2, "item" : "jkl", "price" : NumberDecimal("20"), "quantity" : NumberInt("1"), "date" : ISODate("2014-03-01T09:00:00Z") },
  { "_id" : 3, "item" : "xyz", "price" : NumberDecimal("5"), "quantity" : NumberInt( "10"), "date" : ISODate("2014-03-15T09:00:00Z") },
  { "_id" : 4, "item" : "xyz", "price" : NumberDecimal("5"), "quantity" :  NumberInt("20") , "date" : ISODate("2014-04-04T11:21:39.736Z") },
  { "_id" : 5, "item" : "abc", "price" : NumberDecimal("10"), "quantity" : NumberInt("10") , "date" : ISODate("2014-04-04T21:23:13.331Z") },
  { "_id" : 6, "item" : "def", "price" : NumberDecimal("7.5"), "quantity": NumberInt("5" ) , "date" : ISODate("2015-06-04T05:08:13Z") },
  { "_id" : 7, "item" : "def", "price" : NumberDecimal("7.5"), "quantity": NumberInt("10") , "date" : ISODate("2015-09-10T08:43:00Z") },
  { "_id" : 8, "item" : "abc", "price" : NumberDecimal("10"), "quantity" : NumberInt("5" ) , "date" : ISODate("2016-02-06T20:20:13Z") },
])
db.sales.find()

语法格式

>db.COLLECTION_NAME.aggregate(AGGREGATE_OPERATION)

db.collection.aggregate(
[{
$group: {"_id": "$<分组字段名称>",<显示结果的名称>:{<操作符>: "$<计算的字段>"}}
}
])

# select item, count( quantity ) from sales group by item 
db.sales.aggregate(
[{$group:{"_id" : "$item", "sum":{$sum :"$quantity"} } }
 ])

db.sales.aggregate([{$group: {"_id": null, "count":{$sum: 1}}}]) //8
db.sales.find().size()

2 显示字段 $project

文档字段较多, 只想查询几个字段, 将预显示的字段设定为 “1”;  其余字段默认不显示(也可以设置为”0”), “_id” 较为特别, 默认显示

# 常规显示操作
db.sales.aggregate({"$project": {"_id":0, "item":1}})


# 搭配其他操作符
语法
db.collection.aggrate([
{ $project: {
 <显示结果的名称1>:{<操作符1> : <操作符条件2>}。
 <显示结果的名称2>:{<操作符2>:<操作符条件2>}
}  
}
])

# $substr 需要三个参数：字符串, 起始参数(0 代表第一个字符), 长度参数(如为负表示截取到结尾)
db.sales.aggregate({"$project": {"_id":0, "item_xxxx": {$substr: ["$item", 0,1]}}})

# $switch 对指定字段进行一系列条件判断,符合则返回

db.sales.aggregate({[{"$project"：
{"priceSwitch": {$switch: {branches: [
{case: {$gt: ["$price", 5], then: "price>5"}},
{case: {$lt: ["$price", 10], then: "price<10"}}], 
default:"price <=5 & price >=10"}} 
}
}])

3 数据排序/跳过限制文档数量 $sort $skip $limit

需要对报表进行分页, 将”$sort” “skip” “$limit” 一起使用

/*
SELECT date,
       Sum(( price * quantity )) AS totalSaleAmount,
       Avg(quantity)             AS averageQuantity,
       Count(*)                  AS Count
FROM   sales
GROUP  BY Date(date)
ORDER  BY totalSaleAmount DESC
 */

db.sales.aggregate([
  // First Stage
  {
    $match : { "date": { $gte: new ISODate("2014-01-01"), $lt: new ISODate("2015-01-01") } }
  },
  // Second Stage
  {
    $group : {
       _id : { $dateToString: { format: "%Y-%m-%d", date: "$date" } },
       totalSaleAmount: { $sum: { $multiply: [ "$price", "$quantity" ] } },
       averageQuantity: { $avg: "$quantity" },
       count: { $sum: 1 }
    }
  },
  // Third Stage
  {
    $sort : { totalSaleAmount: -1 }
  }
 ])

5 多表关联查询 $lookup

找出集合中与另一个集合条件匹配的文档, 类似关系型数据库的join

From 需要关联另一个集合， 
localField: 集合中需关联的键， foreignField 与另一个集合关联的键
As  关联后另外一个集合的数据嵌入至此字段下

db.orders.drop()
db.orders.insert([
   { "_id" : 1, "item" : "almonds", "price" : 12, "quantity" : 2 },
   { "_id" : 2, "item" : "pecans", "price" : 20, "quantity" : 1 },
   { "_id" : 3  }
])

// $lookup，多表关联
/*
SELECT *, inventory_docs FROM orders 
WHERE inventory_docs IN (
SELECT * FROM inventory WHERE sku= orders.item
);
 */
db.orders.aggregate([
   {
     $lookup:
       {
         from: "inventory",
         localField: "item",
         foreignField: "sku",
         as: "inventory_docs"
       }
  }
])

6  计算文档数量 $count

db.inventory.aggregate([{$match:{"qty":{$gte:50}}},{$count: "qty_count"}])

7  展开数组 $unwind

将文档中数组形式的数据拆分成数个文档，如指定字段不存在，则不会进行拆分

语法
db.collecton.aggregate([
{$unwind: {path: <field path>, includeArrayInde:<string>, preserveNullAndEmptyArrays:<boolean>}}
])

Path: 指定要拆分的数组，需要用”$”开头
includeArrayInde: 每个数组在原数组中的位置
preserveNullAndEmptyArrays: 选择是否输出未拆分的文档(如空) 默认为false

例
db.inventory.aggregate( [
   {
     $unwind: { path: "$tags", preserveNullAndEmptyArrays: true }
   },
   {
     $group:
       {
         _id: "$tags",
         averageQty: { $avg: "$qty" }
       }
   },
   { $sort: { "averageQty": -1 } }
] )

db.inventory.aggregate( [
   {
     $unwind: { path: "$tags", preserveNullAndEmptyArrays: true }
   },
   {
     $group:
       {
         _id: "$tags",
         averageQty: { $avg: "$qty" }
       }
   },
   { $sort: { "averageQty": -1 } },
   { $skip: 2},
   { $limit: 2}
] )

8 $out

将聚合出来的结果写入一个指定的集合中，如果原集合中有索引，写入的文档违反索引，则写入失败

// $out，将聚合结果汇入新表
db.inventory.aggregate( [
   // First Stage
   {
     $unwind: { path: "$tags", preserveNullAndEmptyArrays: true }
   },
   // Second Stage
   {
     $group:
       {
         _id: "$tags",
         averageQty: { $avg: "$qty" }
       }
   },
   // Third Stage
   {
     $sort: { "averageQty": -1 }
   },
   { $out : "tagsAvgQty" }
] )
db.tagsAvgQty.find();

MapReduce

MapReduce操作具有两个阶段：处理每个文档并向每个输入文档发射一个或多个对象的map阶 段，以及reduce组合map操作的输出的阶段。

• Map 通过键值对形式 将相同 key 的文档 保存到 value 数组中，把结果集key value输出给 reduce阶段
• Reduce 根据 key 对value 进行计算，然后再输出

语法

db.collection.mapreduce(
<map>,  # javascript函数,文档拆分成key  value
<reduce>, # key value进行计算
{
Out: <collection>,
Query: <document>,  #查询/筛选条件
Limit: <number>,     #限制数量 
Finalize: <function>,  # 修改reduce结果然后输出
Scope: <document>,   # 指定map,reduce,finalize函数使用全局变量
jsMode: <boolean>,    #是否在mapreduce过程中将数据转成bson格式
verbose: <boolean>,    #是否在结果中显示时间
bypassDocumentValidation:<boolean>  #选择是否略过数据校验

}
)

db.inventory.find().size()
db.inventory.find()

// 示例
var mapFun = function() {
	// 类似hashmap的put方法，不一样的地方在于同一个Key不是替换，而是追加
   emit(this.status, this.qty);
};
var reduceFun = function(keyCustId, valuesQty) {
  return Array.sum(valuesQty);
};

db.inventory.mapReduce(
 mapFun,
 reduceFun,
 { out: "map_reduce_inventory" }
)


db.map_reduce_inventory.find()