MongoDB 聚合操作

在MongoDB中，有兩種方式計算聚合：Pipeline 和 MapReduce。Pipeline查詢速度快于MapReduce，但是MapReduce的強大之處在于能夠在多臺Server上并行執行復雜的聚合邏輯。MongoDB不允許Pipeline的單個聚合操作占用過多的系統內存，如果一個聚合操作消耗20%以上的內存，那么MongoDB直接停止操作，并向客戶端輸出錯誤消息。

一，使用 Pipeline 方式計算聚合

Pipeline 方式使用db.collection.aggregate()函數進行聚合運算，運算速度較快，操作簡單，但是，Pipeline方式有兩個限制：單個聚合操作消耗的內存不能超過20%，聚合操作返回的結果集必須限制在16MB以內。

創建示例數據，在集合 foo中插入1000條doc，每個doc中有三個field：idx，name 和 age。

for(i=0;i<10000;i++) { db.foo.insert({"idx":i,name:"user "+i,age:i%90})； }

1，使用$match 管道符過濾collection中doc，使符合條件的doc進入pipeline，能夠減少聚合操作消耗的內存，提高聚合的效率。

db.foo.aggregate({$match:{age:{$lte:25}}})

2，使用$project 管道符，使用doc中的部分field進入下級pipeline

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,idx:1,"_id":0}} )

$project 管道符的作用是選擇字段，重命名字段，派生字段。?

2.1 選擇字段

在$project 管道符中，field:1/0，表示選擇/不選擇 field；將無用的字段從pipeline中過濾掉，能夠減少聚合操作對內存的消耗。

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,idx:1,"_id":0}} )

2.2 對字段重命名，產生新的字段

引用符$，格式是："$field"，表示引用doc中 field 的值，如果要引用內嵌 doc中的字段，使用 "$field1.filed2"，表示引用內嵌文檔field1中的字段：field2的值。

示例，新建一個field：preIdx，其值和idx 字段的值是相同的。

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":"$idx",idx:1,"_id":0}} )

2.3 派生字段

在$project中，對字段進行計算，根據doc中的字段值和表達式，派生一個新的字段。

示例，preIdx是根據當前doc的idx 減1 得到的

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":{$subtract:["$idx",1]},idx:1,"_id":0}} )

在$project 執行算術運算的操作符：+($add)，*（$multiply），/（$divide），%（$mod），-（$subtract）。

對于字符數據，$substr:[expr,start,length]用于求子字符串；$concat：[expr1,expr2,,,exprn]，用于將表達式連接在一起；$toLower：expr 和 $toUpper：expr用于返回expr的小寫或大寫形式。

2.4 分組操作

使用$group將doc按照特定的字段的值進行分組，$group將分組字段的值相同的doc作為一個分組進行聚合計算。如果沒有$group 管道符，那么所有doc作為一個分組。對每一個分組，都能根據業務邏輯需要計算特定的聚合值。分組操作和排序操作都是非流式的運算符，流式運算符是指：只要有新doc進入，就可以對doc進行處理，而非流式運算符是指：必須等收到所有的文檔之后，才能對文檔進行處理。分組運算符的處理方式是等接收到所有的doc之后，才能對doc進行分組，然后將各個分組發送給pipeline的下一個運算符進行處理。

示例，按照age進行分組，統計每個分組中的doc數量

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":{$subtract:["$idx",1]},idx:1,"_id":0}} , {$group:{"_id":"$age",count:{$sum:1}}} )

如果分組字段有多個，按照 age 和 age2 進行分組，這樣做僅僅是為了演示，在實際的產品環境中，可以使用更多的字段用來分組。

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":{$subtract:["$idx",1]},idx:1,"_id":0}} , {$group:{"_id":{age:"$age",age2:"$age"},count:{$sum:1}}} )

對每個分組進行聚合運算，count字段是計算每個分組中doc的數量，idxTotal字段是計算每個分組中idx字段值的加和，idxMax字段是計算每個分組中idx字段值的最大值，idxFirst是計算每個分組中第一個idx 字段的值，不一定是最小的。

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":{$subtract:["$idx",1]},idx:1,"_id":0}} , {$group:{"_id":{age:"$age",age2:"$age"},count:{$sum:1},idxTotal:{$sum:"$idx"}},idxMax:{$max:"$idx"},idxFirst:{$first:"$idx"}} } )

2.5，sort操作，limit操作 和 skip操作
對聚合操作的結果進行排序，然后跳過前10個doc，取剩余結果集的前10個doc。

db.foo.aggregate( {$match:{age:{$lte:25}}}, {$project:{age:1,"preIdx":{$subtract:["$idx",1]},idx:1,"_id":0}} , {$group:{"_id":{age:"$age",age2:"$age"},count:{$sum:1},idxTotal:{$sum:"$idx"}},idxMax:{$max:"$idx"},idxFirst:{$first:"$idx"}} }， {$sort:{age:-1}}, {$skip:10}, {$limit:10} )

二，使用MapReduce 方式計算聚合
MapReduce 能夠計算非常復雜的聚合邏輯，非常靈活，但是，MapReduce非常慢，不應該用于實時的數據分析中。MapReduce能夠在多臺Server上并行執行，每臺Server只負責完成一部分wordload，最后將wordload發送到Master Server上合并，計算出最終的結果集，返回客戶端。

MapReduce分為兩個階段:Map和Reduce，舉個例子說明，有10節車廂，統計這10節車廂中男生和女生的數量。串行方式一節一節車廂的統計，直到統計完全部車廂中的人數：男50人，女40人。

使用MapReduce方式的思路是：每個車廂派一個人去統計，每個人返回一個doc，例如，keyN：{female：num1，male：num2}，keyN是車廂編號，在同一時間，有10個人在同時工作，每個人只完成全部workload的10%，很快，返回10個doc，從Key1到Key10，只需要將這10個doc中 femal 和 male分別加和到一起，就是全部車廂的人數：男50人，女40人。

?

使用MapReduce方式計算聚合，主要分為三步：Map，Shuffle和Reduce，Map和Reduce需要顯式定義，shuffle由MongoDB來實現。

• Map：將操作映射到每個doc，產生Key和Value，例如，Map一個doc，產生(female，{count：1})，female是Key，value是{count：1}
• Shuffle：按照Key進行分組，并將key相同的Value組合成數組，例如，產生(female,[{count:1},{count:1},{count:1},{count:1},,,,,])
• Reduce：把Value數組化簡為單值，例如，產生(femal，{count：21})

使用MapReduce進行聚合運算的最佳方式是聚合運算的結果能夠加到一起，例如，求最大值/最小值，sum，平均值（轉換為計算每臺Server的 總和sum1，sum2，，，sumN 與 num1，num2，，numN，平均值avg=（sum1+sum2+，，，+sumN）/（num1+num2+，，+numN））等。

示例，使用MapReduce模擬Count，統計集合中的doc的數量

step1，定義Map函數和reduce函數

對于每個doc，直接返回key 和 一個doc：{count:1}

map=function (){ for(var key in this) {emit(key,{count:1}); } }reduce=function (key,emits){ total=0; for(var i in emits){total+=emits[i].count; } return {"count":total}; }

step2，執行MapReduce運算
在集合 foo上執行MapReduce運算，返回mr 對象

mr=db.runCommand( { "mapreduce":"foo", "map":map, "reduce":reduce, out:"Count Doc" })

step3，查看MapReduce計算的結果

db[mr.result].find()

示例2，統計集合foo中不同age的數量

step1，定義Map 和 Reduce函數

Map函數的作用是對每個doc進行一次映射，返回age 和 {count：1}；

經過Shuffle，每個age都有一個列表：[{count:1},{count:1},{count:1},{count:1},,,,,]，有多少個不同的age，MongoDB都會調用多少次Reduce函數，每次調用時，Key值是不同的。

Reduce函數的作用：對MongoDB的一次調用，對age對應的列表進行聚合運算。

map=function () { emit(this.age,{count:1}); }reduce= function (key,emits) { total=0; for(var i in emits) {total+=emits[i].count; }return {"age":key,count:total}; }

step2，執行MapReduce聚合運算

mr=db.runCommand( { "mapreduce":"foo", "map":map, "reduce":reduce, out:"Count Doc" })

step3，查看聚合運算的結果

db[mr.result].find()

示例3，研究reduce函數的特性

reduce函數具有累加的特性，通過多次調用，能夠產生最終的累加值，例如，以下reduce函數對于任意一個特定的key，reduce都能計算key的數量

reduce= function (key,emits) { total=0; for(var i in emits) {total+=emits[i].count; }return {"key":key,count:total}; }

調用示例：傳遞的Key是相同的，都是“x”，每個emits都是一個數組，反復調用reduce函數，最終獲得key的累加值。

r1=reduce("x",[{count:1},{count:2}]) r2=reduce("x",[{count:3},{count:5}]) r3=reduce("x",[r1,r2])

參考doc：

Aggregation

----@悅光陰 業精于勤而荒于嬉,行成于思而毀于隨 《新程序員》：云原生和全面數字化實踐50位技術專家共同創作，文字、視頻、音頻交互閱讀

總結

以上是生活随笔為你收集整理的MongoDB 聚合操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。