在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])

MongoDB 聚合操作(轉)