下面将会讨论MongoDB中的聚合查询，聚合查询是以管道的方式来完成的，这和Java8中的Stream非常类似，首先聚合查询好比下图流程所示

聚合查询流程

)

管道概念

具体的思路是把Collection加载到聚合框架中，之后经过第一个条件限制（如:math）之后会把筛选出来的数据传给第二个条件，这其实就和管道的概念一样。常用的聚合管道有如下一些部分：

聚合管道的写法如下所示

db.collection.aggregate([{$project},{$match},....{$sort}])

注意写法，首先aggregate是聚合函数，和find一样使用()来调用，第一个参数就是管道，以数组的方式来定义多个管道所以要使用中括号[ ]来定义，数组中的每一个{}就表示一个管道，下面我们会根据上一讲的例子来详细分析管道的各个操作。由于原来的数据库被不小心删除了，下图是最新版本的数据对象模型图，数据结构没有任何的改变，无非一些名称有些改变而已，org变成了dep之类

聚合查询数据模型

)

先来查询一个可以使用find就完成的操作，我们查询年龄大于23的所有User并且仅仅显示name和age，此时需要通过两个管道来完成，match和project，match用来进行条件匹配，project用来确定查询出来的文档

db.user.aggregate([
        {$match:{age:{$gt:23}}},
        {$project:{_id:0,name:1,age:1}}
])

第一个管道完成了条件的查询，第二个管道确定了输出内容。很明显这个操作使用find更为直观一些，接下来将会介绍一些常用了聚合操作，通过聚合操作大家会感受到MongoDB的强大之处。

project和match管道

match用来进行条件筛选，但是注意，该管道放置的位置不一样，筛选所面对的集合是不一样的，我们先来看下面一种查询需求，在设计时，在dep中加入了用户的信息，如果我们希望查询所有的用户数量，使用find就有些力不从心了，需要使用到javascript的shell

var d = db.dep.findOne()
d.users.length

这个还没有办法把所有的信息列表出来，但是使用聚合查询可以很容易实现这个需求

db.dep.aggregate([{$project:{name:1,usize:{$size:"$users"}}}])
{
    "_id" : ObjectId("5a2947c367732445e8adfff0"),
    "name" : "教务处",
    "usize" : 3
}
{
    "_id" : ObjectId("5a2947c367732445e8adfff1"),
    "name" : "财务处",
    "usize" : 3
}
{
    "_id" : ObjectId("5a2947c367732445e8adfff2"),
    "name" : "计算机学院",
    "usize" : 5
}

改查询使用了一个管道，$project，用来投影具体需要的值，这里仅仅显示了name，usize这个可以自己定义，通过 $size 来统计，统计的地方，如果是文档中的key，需要使用$ 符号来引用。

下面加入match，首先通过第一个管道来筛选部门类型为行政部门的信息，看看查询代码

db.dep.aggregate([
    {$match:{type:"行政部门"}},
    {$project:{name:1,unum:{$size:"$users"}}}
])
{
    "_id" : ObjectId("5a2947c367732445e8adfff0"),
    "name" : "教务处",
    "unum" : 3
}
{
    "_id" : ObjectId("5a2947c367732445e8adfff1"),
    "name" : "财务处",
    "unum" : 3
}

第一个管道是基于原来的dep文档来进行筛选的，筛选了type为行政部门的所有信息，接下来我们在project管道之后再使用match来进行筛选，此时就是针对以上文档的，我们需要根据上面文档的值进行查询

db.dep.aggregate([
    {$match:{type:"行政部门"}},
    {$project:{name:1,unum:{$size:"$users"}}},
    {$match:{unum:{$gt:3}}}
])##查询人数大于3的用户
Fetched 0 record(s) in 0ms ##由于没有这个数据，所以显示没有查询到元素

这种查询方式对于关系数据库而言，还是不太容易实现的，需要用到子查询，对于project而言，还可以对文档进行重塑，先来看一个基本的重塑。

db.dep.aggregate([
    {$project:{_id:0,dep:{name:"$name",type:"$type"},unum:{$size:"$users"}}}
])
{
    "dep" : {
        "name" : "教务处",
        "type" : "行政部门"
    },
    "unum" : 3
}

我们把dep中的id去掉，并且把name和type加入了一个子文档dep中，在这个例子中，这种方法似乎比较多余，但是随着我们应用的深入，你会喜欢上这样的操作的，另外MongoDB提供了一组函数来帮助我们做各种重塑，会在后面专门讲解。

group管道

group对于关系型数据库而言，是非常重要的，是统计查询中的基础，MongoDB和关系数据库类似，都是通过group来进行分组查询，group的写法是固定的，{$group:{_id:'分组的对象',多个自定义投影}} 。_id这个是必须填写的，就等于sql中group by 后的这个值，而要统计哪些值，在后面通过写出来，这其实就等于select的投影

下面我们来使用一下group管道，虽然我们设计了在dep下添加users的array，但是我们也在user中添加了dep的映射，我们通过dep来分组统计一下部门下面来看使用group进行统计，这个需要在user这个document中查询

db.user.aggregate([{$group:{_id:'$dep.name',count:{$sum:1}}}])
{  "_id" : "教务处","count" : 3.0}
{"_id" : "财务处","count" : 3.0}
{"_id" : "计算机学院","count" : 5.0}

查询出来的结果是id和数量，使用$group，必须第一个值是_id,说明根据dep的 name进行分组，第二个参数的投影名称是count然后使用$sum ，sum表示求和,等于sql的count(*)。

下面我们希望查询人数大于3的学院，此时在sql中需要使用having来查询，但是对于MongoDB而言，只要在后面增加一个match的管道即可。

db.user.aggregate([
    {$group:{_id:'$dep.name',count:{$sum:1}}},
    {$match:{count:{$gt:3}}}
])
{
    "_id" : "计算机学院",
    "count" : 5.0
}

同样如果希望筛选group前的数据，只用在group管道前增加match即可

db.user.aggregate([
    {$match:{"dep.name":"财务处"}},
    {$group:{_id:'$dep.name',count:{$sum:1}}}
])
{
    "_id" : "财务处",
    "count" : 3.0
}

对于group而言，处理$sum 函数外还有如下几个函数:

以上这些操作很多从字面意思就能知道怎么使用，但是first,last,push和addToSet稍微有些不好理解，我们先通过一个实例把几个统计函数学会，为了方便统计，首先对user加入salary这个数据，并且随机生成薪水的数量,薪水的数量随机4000-10000

Random.setRandomSeed()
db.user.find({}).forEach(function(u){
    db.user.updateOne({_id:u._id},{$set:{salary:(Random.randInt(6000)+4000)}})
})

首先统计每个部门的薪水之和平均薪水，最高薪水和最低薪水

db.user.aggregate([
    {$group:{_id:"$dep.id",
        salays:{$sum:"$salary"},
        avgs:{$avg:"$salary"},
        max:{$max:"$salary"},
        min:{$min:"$salary"}}}
])
{
    "_id" : ObjectId("5a2947c367732445e8adfff0"),
    "salays" : 19394.0,
    "avgs" : 6464.66666666667,
    "max" : 8926.0,
    "min" : 4927.0
}
{
    "_id" : ObjectId("5a2947c367732445e8adfff1"),
    "salays" : 27166.0,
    "avgs" : 9055.33333333333,
    "max" : 9664.0,
    "min" : 8591.0
}

上面这个例子中，根据部门统计了薪水的信息，此时我们由于使用了dep.id进行统计，所以仅仅显示了dep的id，但是我们已经冗余了dep.name，所以可以通过$first 或者$last 函数，由于dep的name是唯一的，所以使用first或者last都一样

db.user.aggregate([
    {$group:{_id:"$dep.id",
        depname:{$first:"$dep.name"},
        salays:{$sum:"$salary"},
        avgs:{$avg:"$salary"},
        max:{$max:"$salary"},
        min:{$min:"$salary"}}}
])
{
    "_id" : ObjectId("5a2947c367732445e8adfff0"),
    "depname" : "教务处",
    "salays" : 19394.0,
    "avgs" : 6464.66666666667,
    "max" : 8926.0,
    "min" : 4927.0
}

最后就是addToSet和push了，这个可以把用户的基本信息添加到一个数组中，使用addToSet添加的信息不会有重复，而使用push的信息会存在重复，看如下的例子

db.user.aggregate([
    {$group:{_id:"$dep.name",
        count:{$sum:1},
        usernames:{$push:'$name'}, #把用户名添加进行
        users:{$addToSet:{phone:'$phone',age:'$age'}}##使用重塑设置两个值
    }}
])

以上这个查询由于没有重复的用户，所以push和addToSet基本一致，但是addToSet中进行了重塑，看看下面的结果

{
    "_id" : "教务处",
    "count" : 3.0,
    "usernames" : [ 
        "b1", 
        "b2", 
        "b3"
    ],
    "users" : [ 
        {
            "phone" : "119",
            "age" : 26
        }, 
        {
            "phone" : "119",
            "age" : 24
        }, 
        {
            "phone" : "119",
            "age" : 23
        }
    ]
}

有了两个数组，一个是usernames，所有用户的名称，users里面是一个对象数组集合。这个数据是非常有必要的，因为可以通过这个信息再次进行二次查询，到此为止group的知识就算讲完了，group是聚合中最为重要的一部分内容，务必掌握。

unwind和out管道

第一个查询需求是获取用户对私人信息的统计，统计每个用户该查询的私人信息数量，看过的数量和没有看过的数量，这使用project就可以解决，先看看结果

db.user.aggregate({$project:
    {
        username:1,
        msgs:{$size:'$msgs.all'},
        noVisited:{$size:'$msgs.noVisited'},
        visited:{$size:'$msgs.visited'}
     }})

但是执行之后会发现报错了，错误提示The argument to $size must be an array, but was of type: missing 提示$size 应该是针对所有的数组，我们的noVisited或者visited可能会存在不存在的情况，这个时候需要通过$ifNull来解决

db.user.aggregate({$project:
   {
       username:1,
       msgs:{$size:'$msgs.all'},
       noVisited:{$size:{$ifNull:['$msgs.noVisited',[]]}},
       visited:{$size:{$ifNull:['$msgs.visited',[]]}}
    }})

注意$ifNull，有两个值，使用的[]而不是{},第一个是条件，第二个是如果不存在的替换值。

看看结果

/* 1 */
{
    "_id" : ObjectId("5a29467b67732445e8adffe5"),
    "username" : "foo1",
    "msgs" : 2,
    "noVisited" : 1,
    "visited" : 1
}

/* 2 */
{
    "_id" : ObjectId("5a2946a567732445e8adffe6"),
    "username" : "bar1",
    "msgs" : 2,
    "noVisited" : 2,
    "visited" : 0
}

下面我们要更进一步，我们希望统计每条私人信息中没有访问的数量，由于访问的信息是存储在user中，并且是以数组的方式来存储，这样很难进行统计，但是MongoDB提供了unwind来解决此类数组查询的问题，使用unwind会把数组中的每个文档都提出来作为一个单独的文档，注意unwind一般和project配合使用，否则会比较占用内存。先看unwind的结果

db.user.aggregate([
    {$project:{name:1,"msgs.noVisited":1}},
    {$unwind:"$msgs.noVisited"}
])
{
    "_id" : ObjectId("5a29467b67732445e8adffe5"),
    "name" : "foo",
    "msgs" : {
        "noVisited" : ObjectId("5a294ee567732445e8adfff6")
    }
}
{
    "_id" : ObjectId("5a2946a567732445e8adffe6"),
    "name" : "bar",
    "msgs" : {
        "noVisited" : ObjectId("5a294ee567732445e8adfff3")
    }
}

大家发现没有，通过unwind，数组中的每一个元素都变成了一个独立的文档，之后通过_id进行一个group即可获取数量

db.user.aggregate([
    {$project:{name:1,"msgs.noVisited":1}},
    {$unwind:"$msgs.noVisited"},
    {$group:{_id:"$msgs.noVisited",count:{$sum:1}}}
])
{
    "_id" : ObjectId("5a29470667732445e8adffef"),
    "count" : 4.0
}
{
    "_id" : ObjectId("5a2946fc67732445e8adffed"),
    "count" : 4.0
}

此时如果希望把标题查询出来，对于MongoDB而言，由于没有join，所以需要对这个数组进行二次查询，使用out可以将查询出来的结果添加到一个文档中

db.user.aggregate([
    {$project:{name:1,"msgs.noVisited":1}},
    {$unwind:"$msgs.noVisited"},
    {$group:{_id:"$msgs.noVisited",count:{$sum:1}}},
    {$out:"msgsNoVisited"}
])

此时可以通过msgsNoVisited来获取message的title，由于MongoDB没有join，所以第一种方法使用游标，利用forEach处理

db.noVisitedMsgs.remove({})
db.msgsNoVisited.find().forEach(function(e) {
    var m = db.message.findOne({_id:e._id})
    if(m!=null) {
        e.title = m.title
    } else {
        e.title = "not found"
    }
    db.noVisitedMsgs.insertOne(e)
})
db.noVisitedMsgs.find()

通过以上shell将查询的结果存储到一个noVisitedMsgs的集合中，第一条remove是先把这个临时数据删除，看一下结果

{
    "_id" : ObjectId("5a294ee567732445e8adfff5"),
    "count" : 9.0,
    "title" : "msg3"
}
{
    "_id" : ObjectId("5a294ee567732445e8adfff4"),
    "count" : 10.0,
    "title" : "msg2"
}

这种方案是基于伪查询的方式，由于每个游标都是通过findOne来进行二次查询的，所以如果数据量非常大的话，效率会很低，在MongoDB3.2版本之后提供了新的函数lookup来完成两个集合的连接

db.message.aggregate([
    {$lookup:{
         from:"msgsNoVisited",
         localField:"_id",
         foreignField:"_id",
         as:"noVisited"
    }}
])

此时会内嵌一个文档noVisited到message中，如下所示

{
    "_id" : ObjectId("5a294ee567732445e8adfff3"),
    "title" : "msg1",
    "content" : "msg ...",
    "createDate" : ISODate("2017-12-15T05:22:20.103Z"),
    "user" : {
        "id" : ObjectId("5a29467b67732445e8adffe5"),
        "name" : "foo"
    },
    "attach" : {
        "name" : "world.txt",
        "type" : "txt",
        "createDate" : "Thu Dec 07 2017 23:34:54 GMT+0800",
        "size" : 23
    },
    "noVisited" : [ 
        {
            "_id" : ObjectId("5a294ee567732445e8adfff3"),
            "count" : 10.0
        }
    ]
}

如果感觉里面的数据太多了，可以首先通过project来进行投影，最后把生成的信息拷贝到一个文档中

db.message.aggregate([
    {$project:{
        title:1
     }},
    {$lookup:{
         from:"msgsNoVisited",
         localField:"_id",
         foreignField:"_id",
         as:"noVisited"
    }},
    {$out:"msgsNoVisited"}
])
db.msgsNoVisited.find()
---------------------------------------
{
    "_id" : ObjectId("5a294ee567732445e8adfff3"),
    "title" : "msg1",
    "noVisited" : [ 
        {
            "_id" : ObjectId("5a294ee567732445e8adfff3"),
            "count" : 10.0
        }
    ]
}

以上就是unwind和out的各种用法，希望对大家有所帮助

sort、skip和limit管道

这三个管道，sort用来排序，skip和limit用来进行分页，这个操作和查询操作非常类似，其实是互通的，看一个实例即可学会。

var pageNum = 1
var pageSize = 10
db.user.aggregate([
    {$project:{name:1,age:1}},
    {$skip:(pageNum-1)*pageSize},
    {$limit:pageSize},
    {$sort:{age:-1}}
])

以上实例通过分页的方式来查询，并且以age倒序排序。

重塑文档

在第一小部分已经看了如何重塑文档，MongoDB还提供了很多好用的函数来重塑文档，重塑文档都是基于project的，这种函数分成很多类，这里只能简单演示一些，首先字符串相关的函数有

字符串相关函数
$concat   连接两个字符串
$strcasecmp 区分大小写比较数字
$substr 取子串
$toLower 转化为小写
$toUpper 转换为大写

看看如下实例，获取用户姓名的信息

db.user.aggregate([
    {$project:{
        fullname:{$concat:['$username','-','$name']},
        fname:{$substr:['$username',0,1]}
     }}
])

{
    "_id" : ObjectId("5a29467b67732445e8adffe5"),
    "fullname" : "foo1-foo",
    "fname" : "f"
}

接下来看看算术运算函数和日期函数

算术运算函数
    $add   求和
    $divide 除法
    $mod 求余数
    $multiply 乘法
    $subtract 减法
日期函数
    $year 取年份
    $month 取月份
    $week 取一年中的某一周(0-53)
    $hour 取小时
    $minute 取分钟
    $second 取秒钟
    $millisecond 取毫秒
    $dayOfYear 一年中的某一天
    $dayOfMonth 一月中的某一天
    $dayOfWeek 一周中的某一天，1表示周日

算数运算符比较简单，我们使用以下日期函数

db.message.aggregate([
    {$match:{createDate:{$gte:new Date("2017-01-01"),$lte:new Date("2017-12-31")}}},
    {$project:{title:1,year:{$year:"$createDate"},month:{$month:"$createDate"}}},
    {$group:{_id:{year:'$year',month:'$month'},count:{$sum:1}}}
])

以上操作完成了根据year和month进行分组查询，最后可以得到2017年每个月发送的message的数量，这是非常好用的一种聚合查询。

集合操作函数也非常实用

集合操作符
    $setEquals  查看两个集合时候相同，相同返回true
    $setInterseciont 返回两个集合的公共元素
    $setDifference 返回两个集合中的不同的元素
    $setUnion 合并集合
    $setIsSubset 判断第二个集合是否是第一个的子集，如果是返回true
    $anyElementTrue 如果某个集合元素为true，就为true
    $allElementTrue 集合中所有元素为true就为true

看一下下一个实例

db.user.aggregate([
    {$project:{nov:{$setDifference:['$msgs.all',{$ifNull:['$msgs.visited',[]]}]}}}
])

会得到没有参观过的message信息，下面看一下其他的重塑函数

逻辑函数
    $and true   与操作
    $cmp 比两个值是否相等
    $cond  if ... then ... else 条件逻辑，类似三元运算符
    $eq 等于
    $gt,$gte,$lt,$lte  大于和小于
    $ifNull 是否为空
    $ne 不等于
    $not. 取反
    $ir 或运算
其他函数
    $meta  文本搜索
    $size 取数组大小
    $map 对数组的每个成员应用表达式
    $let 定义表达式内的变量
    $literal 返回表达式的值

以上函数就不举例了，只是让大家对这个有所了解，以后具体使用到的时候再去查询

Map-Reduce

对于一些复杂的查询可以转换为javascript的处理方式，这就是map reduce，但是map reduce的效率非常低，所以能够使用聚合框架实现的就不要使用Map-Reduce，下面看看Map-Reduce如何实现一个应用。

首先要定义map函数，map指的是遍历文档中的所有键值对，来做处理，我们现在对user这个collection来做map

map = function() {
    var did = this.dep.id+"-"+this.dep.name;
    var ageGt = 0;
    if(this.age>40) ageGt=1;
    emit(did,{ageGt:ageGt});
}

以上操作定义了一个map，该操作会去遍历整个collection的文档，首先确定了key是部门id+部门名称，接着完成下面一种存储如果年龄大于40存储ageGt为1否则为0，emit表示返回值，此时的map文档中有一组数据，存储了部门id+部门名称为key并且存储了年龄的文档（年龄的文档中一些是0，一些是1），之后把这个值交给reduce来返回一个结果集

reduce=function(key,values) {
    var result = {nums:0};
    values.forEach(function(v){
        result.nums+=v.ageGt;
    })
    return result;
}

reduce中就会根据key的值把返回的数据定义成为一个数组，这个数组就是reduce的第二个参数values，此时定义了一个结果为nums，并且遍历整个数组，将nums的值增加，这意味着只要发现ageGt是1表示有一个40岁以上的人，这样累加之后，就可以根据情况统计出每个部门大于40岁的人员。

定义好map和reduce之后，下一步就需要通过collection来调用

db.user.mapReduce(map,reduce,{query:{},out:"test"})

此时使用user调用刚才定义的map和reduce，第三个参数中的query是查询条件，可以过滤一组数据，第二个参数out表示要输出的文档，输出之后我们可以通过test文档查询

/* 1 */
{
    "_id" : "5a2947c367732445e8adfff0-教务处",
    "value" : {
        "nums" : 0.0
    }
}

/* 2 */
{
    "_id" : "5a2947c367732445e8adfff1-财务处",
    "value" : {
        "nums" : 1.0
    }
}

/* 3 */
{
    "_id" : "5a2947c367732445e8adfff2-计算机学院",
    "value" : {
        "nums" : 0.0
    }
}

这个map-reduce完成了部门大于40岁人的统计操作，Map-Reduce基本可以实现你想要的所有功能，但是由于效率低下，依然建议大家如果能使用聚合框架就不要用MapReduce。

总结

这部分讲解了聚合框架，这是MongoDB的核心，它可以提供非常出色的查询操作，这是必须要熟悉的一块内容，当然聚合框架的效率问题将会在索引之后详细探讨。