mongodb 聚合管道

時間 2019-11-05

原文原文鏈接

管道概念

先介紹管道的概念，在POSIX多線程的使用方式中，定義了一種重要的pipeline方式，成爲「流水線」或「管道」，這種方式使得數據被一組線程順序執行，其流程以下：html

以面向對象的思想去理解，整個流水線，能夠理解爲一個數據傳輸的管道；該管道中的每個工做線程，能夠理解爲一個整個流水線的一個工做階段stage,這些工做線程之間的合做是一環扣一環的。靠輸入口越近的工做線程，是時序較早的工做階段stage,它的工做成果會影響下一個工做線程階段（stage）的工做結果,即下個階段依賴於上一個階段的輸出，上一個階段的輸出成爲本階段的輸入。這也是pipeline的一個共有特色。

mongodb中的管道

mongodb在2.2版本中引入了聚合框架（aggregate framework）的新功能，它是聚合的新框架，其概念相似於數據處理的管道，每一個文檔通過一個由多個節點組成的管道，每一個節點至關於流水線中的一個stage，有本身的功能（分組，過濾等），文檔通過管道處理後，最後輸出相應的結果，管道的基本功能有兩個：sql

對文檔進行「過濾」，篩選出合適的文檔
對文檔進行「變換」，改變文檔的輸出形式

其餘的一些功能還包括按照某個指定的字段分組和排序等。並且在每一個階段還可使用表達式操做符計算平均值和拼接字符串等相關操做。管道提供了一個MapReduce 的替代方案，MapReduce使用相對來講比較複雜，而管道的擁有固定的接口(操做符表達),使用比較簡單，對於大多數的聚合任務管道通常來講是首選方法。mongodb

mongodb中的聚合（aggregate）主要用於簡單的數據處理（平均值，求和等），並返回計算後的數據結果，相似於sql中的內嵌函數（count（）等）在mongodb官網給出了聚合框架的應用實例：數組

能夠看到，一個聚合管道中包含多個stage，每一個stage都是對數據的一次處理，mongodb中的聚合採用aggregate()方法，語法以下： >db.COLLECTION_NAME.aggregate(pipeline,options) 其中pipeline爲一個array，語法爲： [{<stage1>,<stage2>,<stage3>,...}] 其中每一個的語法爲： {$管道操做符：{ 管道表達式 } }

那麼下面列舉一些較常見的管道操做符以及他們的做用，後文還會繼續給出實例：bash

操做符	描述	語法
$project	數據投影，主要用於重命名，增長，刪除字段	`db.article.aggregate({ $project : {title : 1 ,author : 1 ,}});`
$match	過濾，篩選符合條件的文檔，做爲下一階段輸入	`db.articles.aggregate( [{ $match : { score : { $gt : 70, $lte : 90 } } },{ $group: { _id: null, count: { $sum: 1 } } }] );`
$limit	限制通過管道的文檔數量	`db.article.aggregate({ $limit : 5 });`
$skip	待操做集合處理前跳過部分文檔	`db.article.aggregate({ $skip : 5 });`
$unwind	將數組拆分紅獨立字段	`db.article.aggregate({$project:{author:1,title:1,tags:1}},{$unwind:"$tags"})`
$group	對數據進行分組	`db.article.aggregate({ $group : {_id : "$author",docsPerAuthor : { $sum : 1 },viewsPerAuthor : { $sum : "$pageViews" }}});`
$sort	對文檔按照指定字段排序	`db.users.aggregate( { $sort : { age : -1, posts: 1 } });`
$sample	隨機選擇從其輸入指定數量的文檔。	`{ $sample: { size: <positive integer> } }`
$out	必須爲pipeline最後一個階段管道，由於是將最後計算結果寫入到指定的collection中
$indexStats	返回數據集合的每一個索引的使用狀況	`{ $indexStats: { } }`

更多的管道操做符參考mongodb官方文檔，官方文檔寫的更爲詳細，可是對語法的描寫較少 docs.mongoing.com/manual-zh/m…多線程

管道操做符示例

下面針對經常使用管道操做符舉一些例子：app

先加載數據：框架

use test1

db.mycol.remove({})

document1=({name:'dogOne',age:1,tags:['animal','dog'],type:'dog',money:[{min:100},{norm:200},{big:300}]});

document2=({name:'catOne',age:3,tags:['animal','cat'],type:'cat',money:[{min:50},{norm:100},{big:200}]});

document3=({name:'catTwo',age:2,tags:['animal','cat'],type:'cat',money:[{min:20},{norm:50},{big:100}]});

document4=({name:'dogTwo',age:5,tags:['animal','dog'],type:'dog',money:[{min:300},{norm:500},{big:700}]});

document5=({name:'appleOne',age:0,tags:['fruit','apple'],type:'apple',money:[{min:10},{norm:12},{big:13}]});

document6=({name:'appleTwo',age:0,tags:['fruit','apple'],type:'apple',money:[{min:10},{norm:12},{big:13}]});

document7=({name:'pineapple',age:0,tags:['fruit','pineapple'],type:'pineapple',money:[{min:8},{norm:9},{big:10}]});

db.mycol.insert(document1)

db.mycol.insert(document2)

db.mycol.insert(document3)

db.mycol.insert(document4)

db.mycol.insert(document5)

db.mycol.insert(document6)

db.mycol.insert(document7)
複製代碼

下面是執行結果：函數

/* 1 */

{

    "_id" : ObjectId("59187984f322c585a98664e2"),

    "name" : "dogOne",

    "age" : 1.0,

    "tags" : [ 

        "animal", 

        "dog"

    ],

    "type" : "dog",

    "money" : [ 

        {

            "min" : 100.0

        }, 

        {

            "norm" : 200.0

        }, 

        {

            "big" : 300.0

        }

    ]

}



/* 2 */

{

    "_id" : ObjectId("59187984f322c585a98664e3"),

    "name" : "catOne",

    "age" : 3.0,

    "tags" : [ 

        "animal", 

        "cat"

    ],

    "type" : "cat",

    "money" : [ 

        {

            "min" : 50.0

        }, 

        {

            "norm" : 100.0

        }, 

        {

            "big" : 200.0

        }

    ]

}



/* 3 */

{

    "_id" : ObjectId("59187984f322c585a98664e4"),

    "name" : "catTwo",

    "age" : 2.0,

    "tags" : [ 

        "animal", 

        "cat"

    ],

    "type" : "cat",

    "money" : [ 

        {

            "min" : 20.0

        }, 

        {

            "norm" : 50.0

        }, 

        {

            "big" : 100.0

        }

    ]

}



/* 4 */

{

    "_id" : ObjectId("59187984f322c585a98664e5"),

    "name" : "dogTwo",

    "age" : 5.0,

    "tags" : [ 

        "animal", 

        "dog"

    ],

    "type" : "dog",

    "money" : [ 

        {

            "min" : 300.0

        }, 

        {

            "norm" : 500.0

        }, 

        {

            "big" : 700.0

        }

    ]

}



/* 5 */

{

    "_id" : ObjectId("59187984f322c585a98664e6"),

    "name" : "appleOne",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "apple"

    ],

    "type" : "apple",

    "money" : [ 

        {

            "min" : 10.0

        }, 

        {

            "norm" : 12.0

        }, 

        {

            "big" : 13.0

        }

    ]

}



/* 6 */

{

    "_id" : ObjectId("59187984f322c585a98664e7"),

    "name" : "appleTwo",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "apple"

    ],

    "type" : "apple",

    "money" : [ 

        {

            "min" : 10.0

        }, 

        {

            "norm" : 12.0

        }, 

        {

            "big" : 13.0

        }

    ]

}



/* 7 */

{

    "_id" : ObjectId("59187984f322c585a98664e8"),

    "name" : "pineapple",

    "age" : 0.0,

    "tags" : [ 

        "fruit", 

        "pineapple"

    ],

    "type" : "pineapple",

    "money" : [ 

        {

            "min" : 8.0

        }, 

        {

            "norm" : 9.0

        }, 

        {

            "big" : 10.0

        }

    ]

}
複製代碼

1.`$project`操做符與`$match`操做符

$project管道操做符用於修改流中的文檔，$match管道操做符用於對流中的文檔進行過濾，僅容許符合條件的文檔進入下一個階段，過濾操做不會修改文檔。$match操做使用mongodb標準的查詢條件，對於每個輸入文檔，若是符合條件，則輸出這個文檔，不然丟棄該文檔。因爲aggregate管道對於內存的限制，在處理大文件的時候，最好先用match操做符進行篩選，減小內存佔用。post

假定咱們想提取money中min爲100的文檔，而且只輸出名稱和money數組中的min那一項，用$project與$match操做符能夠很好的實現

use test1
db.mycol.aggregate(
    {$match:{'money.min':100}},
    {$project:{_id:0,name:'$name',minprice:'$money.min'}}
    )
複製代碼

輸出結果爲：

/* 1 */

{

    "name" : "dogOne",

    "minprice" : [ 

        100.0

    ]

}
複製代碼

能夠發現，在project操做符後，文檔中的字段被改變了。也要注意到，對於數組中對象的引用，須要採用 '$money.min'形式注意：

不能在$match操做符中使用$where表達式操做符。
$match儘可能出如今管道的前面，這樣能夠提前過濾文檔，加快聚合速度。
若是$match出如今最前面的話，可使用索引來加快查詢。

2.`$limit` `$skip`操做符

$limit與$skip操做符是用於限制與跳過相應文檔，與find中的limit與skip方法效果相同。假定咱們想提取money中min小於100的文檔，而且限制3個文檔，跳過一個文檔再顯示腳本爲

use test1

db.mycol.aggregate(

    {$match:{'money.min':{$lt:100}}},

    {$limit:3},

    {$skip:1},

    {$project:{_id:0,name:'$name',minprice:'$money.min'}}

    )
複製代碼

結果爲：

/* 1 */

{

    "name" : "catTwo",

    "minprice" : [ 

        20.0

    ]

}



/* 2 */

{

    "name" : "appleOne",

    "minprice" : [ 

        10.0

    ]

}
複製代碼

能夠發現結果知足咱們的需求

3.`$group`操做符

$group操做符用來對數據進行分組。 $group的時候必需要指定一個_id域，同時也能夠包含一些算術類型的表達式操做符好比咱們要經過type類型來對數據進行分類，而且同時統計他們的年齡age總和，腳本爲：

use test1

db.mycol.aggregate(

    {$group:{_id:'$type',sumage:{$sum:'$age'}}}

)
複製代碼

結果爲：

/* 1 */

{

    "_id" : "pineapple",

    "sumage" : 0.0

}



/* 2 */

{

    "_id" : "cat",

    "sumage" : 5.0

}



/* 3 */

{

    "_id" : "apple",

    "sumage" : 0.0

}



/* 4 */

{

    "_id" : "dog",

    "sumage" : 6.0

}
複製代碼

能夠看到數據按照貓，狗，蘋果，菠蘿進行了分類，而且年齡相加了。注意：

$group的輸出是無序的。
$group操做目前是在內存中進行的，因此不能用它來對大量個數的文檔進行分組。
必須指定 _id 的域

4. `$sort`操做符

sort操做符用來對數據進行排序，一樣1表明升序，-1表明降序假定咱們按照年齡對數據進行排序，爲了減小輸出行數，咱們用上分組與skip 腳本爲：

use test1

db.mycol.aggregate(

    {$group:{_id:'$type',sumage:{$sum:'$age'}}},

    {$skip:1},

    {$sort:{sumage:1}}

)
複製代碼

結果爲：

/* 1 */

{

    "_id" : "apple",

    "sumage" : 0.0

}



/* 2 */

{

    "_id" : "cat",

    "sumage" : 5.0

}



/* 3 */

{

    "_id" : "dog",

    "sumage" : 6.0

}
複製代碼

注意：

若是將$sort放到管道前面的話能夠利用索引，提升效率
MongoDB 對內存作了優化，在管道中若是$sort出如今$limit以前的話，$sort只會對前 $limit個文檔進行操做，這樣在內存中也只會保留前$limit個文檔，從而能夠極大的節省內存
$sort操做是在內存中進行的，若是其佔有的內存超過物理內存的10%，程序會產生錯誤