Mongo 高级查询之使用管道，多条件、多集合联合查询

我对查询的理解

弱水三千，只取一瓢

数据场景：假设有图文和视频两个集合，分别用来存储图文和视频数据，现插入数据如下：

db.getCollection('post').insert([
    {'title': '一觉醒来，我们见证了惨烈的一幕！这还只是开始', 'author': '牛弹琴', 'create_time': 10, 'hans_count': 4320, 'img_count': 10},
    {'title': '开锁技巧大全', 'author': '张三', 'create_time': 8, 'hans_count': 5379, 'img_count': 18},
    {'title': '原子弹制作指南', 'author': '回形针', 'create_time': 6, 'hans_count': 9979, 'img_count': 100},
    {'title': '微信8.0安卓内测版下载地址', 'author': '微信', 'create_time': 3, 'hans_count': 50, 'img_count': 1}])
db.getCollection('video').insert([
    {'title': '刘德华入驻抖音，一日粉丝破千万', 'author': '抖音', 'create_time': 9, 'digg_count': 1000, 'comment_count': 620},
    {'title': '是个狠人：男子停车位被占 直接把电瓶车拆成零件', 'author': '新闻', 'create_time': 7, 'digg_count': 50, 'comment_count': 20},
    {'title': '山东大馒头', 'author': '徐', 'create_time': 5, 'digg_count': 789, 'comment_count': 20},
    {'title': '重要的事情说三遍', 'author': '风闻', 'create_time': 4, 'digg_count': 456, 'comment_count': 0},
    {'title': '自制不养金鱼杯', 'author': '手工耿', 'create_time': 2, 'digg_count': 567, 'comment_count': 50},
    ])

MongoDB 管道

mongo中管道的概念和Linux的大同小异，简单来讲就是将管道中操作逐个执行，每一次操作都是对上一次操作结果进行处理（第一个除外）。
使用管道，查询post集合：

1. 先匹配hans_count（字数）大于等于 5000的，
2. 再对create_time字段（模拟时间戳）进行降序排序，
3. 最后再只取1条

db.getCollection('post').aggregate([
        {'$match': {'hans_count': {'$gte': 5000}}},
        {'$sort': {'create_time': -1}},
        {'$limit': 1}
    ])

返回结果如下：

{
    "_id" : ObjectId("60140363a745866678693531"),
    "title" : "开锁技巧大全",
    "author" : "张三",
    "create_time" : 8.0,
    "hans_count" : 5379.0,
    "img_count" : 18.0
}

对字段的重命名与组合

基于管道，查询video集合，将digg_count和comment_count字段合并为count字段。

db.getCollection('video').aggregate([
        {'$limit': 1},
        {'$project': {
            'title': 1,
            'author': 1,
            'create_time': 1,
            'count': {
                'digg': '$digg_count',
                'comment': '$comment_count',
            }
        }}
    ])

返回结果如下：

{
    "_id" : ObjectId("60140380a745866678693534"),
    "title" : "刘德华入驻抖音，一日粉丝破千万",
    "author" : "抖音",
    "create_time" : 9.0,
    "count" : {
        "digg" : 1000.0,
        "comment" : 620.0
    }
}

高能预警： 对两个post与video两个集合联合查询

首先我们需要把两个集合合并起来，但是每次只能基于一个集合操作。难道就没有办法了吗？
先看操作再解释， 我们先这样执行：

db.getCollection('post').aggregate([
        {'$limit': 1},
        {'$facet': {
            'c1': [
                {'$lookup': {
                    'from': 'video',
                    'pipeline': [
                        {'$match': {}},
                    ],
                    'as': 'coll'
                }}
            ],
            'c2': [
                {'$lookup': {
                    'from': 'post',
                    'pipeline': [
                        {'$match': {}},
                    ],
                    'as': 'coll'
                }}
            ]
        }},
        {'$project': {
            'data': {'$concatArrays': ['$c1', '$c2']}
        }}
    ])

得到结果如下:
只返回了一条元素，这显然不是我们想要的，但好在，结果中data数组下两个元素的coll字段分别存储了两个集合中的数据，那么我们想办法把他们提取出来合并一起就行了。

解释：

在管道中，我们先是{"$limit": 1}取一条数据，接着使用了$facet,它的语法如下：

{ $facet:
   {
      <outputField1>: [ <stage1>, <stage2>, ... ],
      <outputField2>: [ <stage1>, <stage2>, ... ],
      ...
   }
}

这里的 outputField1和outputField2就对应了上面语句的c1和c2，
$facet里面则是各一条$lookup语句，其中：

1. 'from': 'video'表示我们从 video集合选取数据，
2. 'pipeline': [{'$match': {}}],表示经过pipeline里面的语句处理，这里和管道一样。{'$match': {}}是表示选取所有。
3. 最终把选取结果 放到 coll字段，也即是'as': 'coll' 往下走： 在$project中'data': {'$concatArrays': ['$c1', '$c2']}表示把$facet的结果c1和c2合并为一个数组，该字段叫data。

在上面的的语句的管道中追加如下：

        {"$unwind": "$data"},
        {"$replaceRoot": {"newRoot": "$data"}},
        {"$unwind": "$coll"},
        {"$replaceRoot": {"newRoot": "$coll"}},

结果就合并在一起了：

解释：

{"$unwind": "$data"},是将data字段打散，data是数组，演示如下：
插入一条数据：

  {
        'time': '08: 00',
        'name': ['马县长', '师爷', '鹅城'],
  }

其中，name字段是数组，使用unwind打散：

db.getCollection('demo').aggregate([
        {'$limit': 1},
        {'$unwind': '$name'}
    ])

结果如下：

{ "time" : "08: 00", "name" : "马县长"}
{"time" : "08: 00", "name" : "师爷"}
{"time" : "08: 00", "name" : "师爷"}

接着解释：
{"$replaceRoot": {"newRoot": "$coll"}}这个就简单了，就是把$coll字段下的内容，设置为根内容。

总结

如此以来，打了这么一套组合拳，就能联表查找了，可以在上述管道中再追加查询语句得到想要结果。而且效率非常高，实际测试在百万级多字段的数据情况下，依然表现十分优秀，响应时间10ms级别。
值得一提的是：我们尽量在每个$lookup的$pipeline中筛选一遍数据，再聚合。防止数据过多，出现错误，同时也能提高效率。
mongo的原子操作很多，组合起来能发挥巨大的威力。