快速掌握mongoDB(二)——聚合管道和MapReduce
上一節簡單介紹了一下mongoDB的增刪改查操作，這一節將介紹其聚合操作。我們在使用mysql、sqlserver時經常會用到一些聚合函式，如sum/avg/max/min/count等，mongoDB也提供了豐富的聚合功能，讓我們可以方便地進行資料的分析和計算。這裡主要介紹兩種聚合方式：聚合管道和MapReduce.
1 聚合管道
官網文件：https://docs.mongodb.com/manual/core/aggregation-pipeline/
聚合管道(aggregation pipeline)，顧名思義是基於管道模式的聚合框架，簡單的說就是在聚合管道中前一個階段處理的結果會作為後一階段處理的輸入，documents通過管道中的各個階段處理後得到最終的聚合結果。聚合管道的語法： db.aggregate( [ { stage1 },{ stage2} ... ] ) 。
我們看一個官網提供的栗子就理解了：
//準備測試資料
db.orders.insertMany([
{cust_id:"A123",amount:500,status:"A"},
{cust_id:"A123",amount:250,status:"A"},
{cust_id:"B212",amount:200,status:"A"},
{cust_id:"A123",amount:300,status:"D"}
])
//聚合操作
db.orders.aggregate([
{$match:{status:"A"}},
{$group:{_id:"$cust_id",total:{$sum:"$amount"}}}
])執行上邊的命令，結果如下：
上邊的聚合過程一共有兩個階段，如下圖所示：
第一階段：$match會篩選出status="A"的documents，並把篩選出的document傳給下一個階段；
第二階段：$group將上一階段傳過來的documents通過cust_id進行分組，併合計各組的amount。
通過官網的栗子我們大概知道了集合管道的基本執行流程，下邊我們通過幾個栗子來理解幾個常用的聚合運算子。
栗子1：$lookup,$match,$project,$group,$sort,$skip,$limit,$out
首先看一個訂單庫存的栗子，我們將分步檢視各個階段的聚合結果：
////準備測試資料
//訂單
db.orders.insertMany([
{ "_id" : 1, "item" : "almonds", "price" : 12, "quantity" : 2 },
{ "_id" : 2, "item" : "bread", "price" : 8, "quantity" : 3 },
{ "_id" : 3, "item" : "pecans", "price" : 20, "quantity" : 1 },
{ "_id" : 4, "item" : "pecans", "price" : 20, "quantity" : 3 },
{ "_id" : 5, "item" : "cashews", "price" : 25, "quantity" :2},
{ "_id" : 6 }
])
//庫存
db.inventory.insertMany([
{ "_id" : 1, "sku" : "almonds", description: "product 1", "instock" : 120 },
{ "_id" : 2, "sku" : "bread", description: "product 2", "instock" : 80 },
{ "_id" : 3, "sku" : "cashews", description: "product 3", "instock" : 60 },
{ "_id" : 4, "sku" : "pecans", description: "product 4", "instock" : 70 },
{ "_id" : 5, "sku": null, description: "Incomplete" },
{ "_id" : 6 }
])
db.orders.aggregate([
//$lookup實現collection連線，作用類似於sql中的join
{
$lookup:
{
from: "inventory",//要join的集合
localField: "item",//連線時的本地欄位
foreignField: "sku",//連線時的外部欄位
as: "inventory_docs"//連線後新增陣列欄位的名字
}
},
//$match過濾document，結果集合只包含符合條件的集合
{$match:
{price:{$gt:0}}
},
//$project用於獲取結果欄位，可以新增欄位;也可以對已存在的欄位重新賦值
{
$project:{
_id:0,
item:1,
price:1,
quantity:1,
totalprice:{$multiply:["$price","$quantity"]}
}
},
//$group實現分組，_id是必須的，用於指定分組的欄位；這裡查詢各個分組的totalprice的最大值
{
$group:{_id:"$item",maxtotalprice:{$max:"$totalprice"}}
},
//$sort用於排序，1表示正序，-1表示倒序
{$sort:{maxtotalprice:-1}},
//$skip用於跳過指定條數的document，和linq中的skip作用一樣
{$skip:1}，
//limit用於指定傳遞給下一階段的document條數，和mysql的limit作用一樣
{$limit:2},
//$out用於將結果存在指定的collection中，如果collection不存在則新建一個，如果存在則用結果集合覆蓋以前的值
{$out:"resultCollection"}
])
$ lookup階段 用於collection連線，作用類似於sql中的join，經過該階段結果如下：
[
{
"_id" : 1,
"item" : "almonds",
"price" : 12,
"quantity" : 2,
"inventory_docs" : [
{
"_id" : 1,
"sku" : "almonds",
"description" : "product 1",
"instock" : 120
}
]
},
{
"_id" : 2,
"item" : "bread",
"price" : 8,
"quantity" : 3,
"inventory_docs" : [
{
"_id" : 2,
"sku" : "bread",
"description" : "product 2",
"instock" : 80
}
]
},
{
"_id" : 3,
"item" : "pecans",
"price" : 20,
"quantity" : 1,
"inventory_docs" : [
{
"_id" : 4,
"sku" : "pecans",
"description" : "product 4",
"instock" : 70
}
]
},
{
"_id" : 4,
"item" : "pecans",
"price" : 20,
"quantity" : 3,
"inventory_docs" : [
{
"_id" : 4,
"sku" : "pecans",
"description" : "product 4",
"instock" : 70
}
]
},
{
"_id" : 5,
"item" : "cashews",
"price" : 25,
"quantity" : 2,
"inventory_docs" : [
{
"_id" : 3,
"sku" : "cashews",
"description" : "product 3",
"instock" : 60
}
]
},
{
"_id" : 6,
"inventory_docs" : [
{
"_id" : 5,
"sku" : null,
"description" : "Incomplete"
},
{
"_id" : 6
}
]
}
]
View Code
$match階段 用於篩選出符合過濾條件的documents，相當於sql中的where
[
{
"_id" : 1,
"item" : "almonds",
"price" : 12,
"quantity" : 2,
"inventory_docs" : [
{
"_id" : 1,
"sku" : "almonds",
"description" : "product 1",
"instock" : 120
}
]
},
{
"_id" : 2,
"item" : "bread",
"price" : 8,
"quantity" : 3,
"inventory_docs" : [
{
"_id" : 2,
"sku" : "bread",
"description" : "product 2",
"instock" : 80
}
]
},
{
"_id" : 3,
"item" : "pecans",
"price" : 20,
"quantity" : 1,
"inventory_docs" : [
{
"_id" : 4,
"sku" : "pecans",
"description" : "product 4",
"instock" : 70
}
]
},
{
"_id" : 4,
"item" : "pecans",
"price" : 20,
"quantity" : 3,
"inventory_docs" : [
{
"_id" : 4,
"sku" : "pecans",
"description" : "product 4",
"instock" : 70
}
]
},
{
"_id" : 5,
"item" : "cashews",
"price" : 25,
"quantity" : 2,
"inventory_docs" : [
{
"_id" : 3,
"sku" : "cashews",
"description" : "product 3",
"instock" : 60
}
]
}
]
View Code
$project階段 用於設定查詢的欄位，想到於sql中的select field1,field2...
[
{
"item" : "almonds",
"price" : 12,
"quantity" : 2,
"totalprice" : 24
},
{
"item" : "bread",
"price" : 8,
"quantity" : 3,
"totalprice" : 24
},
{
"item" : "pecans",
"price" : 20,
"quantity" : 1,
"totalprice" : 20
},
{
"item" : "pecans",
"price" : 20,
"quantity" : 3,
"totalprice" : 60
},
{
"item" : "cashews",
"price" : 25,
"quantity" : 2,
"totalprice" : 50
}
]
View Code
$group 用於分組，相當於sql中的group by，經過該階段結果如下：
[
{ "_id" : "cashews","maxtotalprice" : 50},
{ "_id" : "pecans","maxtotalprice" : 60},
{ "_id" : "bread", "maxtotalprice" : 24},
{ "_id" : "almonds","maxtotalprice" : 24}
]
$sort階段 用於排序，相當於sql中的 sort by，其中值為1表示正序，-1表示反序，經過該階段結果如下：
[
{ "_id" : "pecans","maxtotalprice" : 60},
{ "_id" : "cashews","maxtotalprice" : 50},
{ "_id" : "bread", "maxtotalprice" : 24},
{ "_id" : "almonds","maxtotalprice" : 24}
]
$skip階段 用於跳過指定條數的document，和linq中的skip作用一樣，經過該階段結果如下：
[
{ "_id" : "cashews", "maxtotalprice" : 50},
{ "_id" : "bread","maxtotalprice" : 24},
{ "_id" : "almonds", "maxtotalprice" : 24}
]
$limit階段 用於指定傳遞給下一階段的document條數，相當於mysql中的limit，和linq中的take作用一樣，經過該階段結果如下：
[
{"_id" : "cashews","maxtotalprice" : 50},
{"_id" : "bread","maxtotalprice" : 24}
]
$out階段，用於將結果存在指定的collection中，如果collection不存在則新建一個，如果存在則用結果集合覆蓋以前的值，這裡我們將 db.resultCollection.find() 檢視resultCollection，結果如下：
[
{ "_id" : "cashews", "maxtotalprice" : 50},
{ "_id" : "bread", "maxtotalprice" : 24}
]
栗子2：$addFields，$unwind,$count
看一個使用者分析的栗子，新增測試資料：
//測試資料
db.userinfos.insertMany([
{ _id:1, name: "王五", age: 25, roles:["vip","gen" ]},
{ _id:2, name: "趙六", age: 26, roles:["vip"]},
{ _id:3, name: "田七", age: 27}
])
$addFields 用於給所有的document新增新欄位，如果欄位名已經存在的話，用新值替代舊值
db.userinfos.aggregate([
{$addFields:{address:'上海',age:30}}
])
執行上邊的命令後，結果如下：
[{"_id":1,"name":"王五","age":30,"roles":["vip","gen"],"address":"上海"},
{"_id":2,"name":"趙六","age":30,"roles":["vip"],"address":"上海"},
{"_id":3,"name":"田七","age":30,"address":"上海"}]
$unwind $unwind用於對陣列欄位解構，陣列中的每個元素都分解成一個單獨的document
db.userinfos.aggregate([
{$unwind:"$roles"}
])
執行命令後，結果如下：
[{"_id" : 1,"name" : "王五","age" : 25,"roles" : "vip"},
{"_id" : 1,"name" : "王五","age" : 25,"roles" : "gen"},
{"_id" : 2,"name" : "趙六","age" : 26,"roles" : "vip"}]
$count 用於獲取document的條數，輸入的值為欄位名字，用法十分簡單
db.userinfos.aggregate([
{$count:"mycount"}
])
執行命令後，結果如下：
[ { "mycount" : 3 } ]
栗子3 $bucket,$bucketAuto,$sample
看一個網上書店的栗子，首先新增測試資料：
db.books.insertMany([
{ "_id" : 1, "title" : "《白鯨》", "artist" : "赫爾曼・梅爾維爾（美）", "year" : 1845,"price" : NumberDecimal("199.99") },
{ "_id" : 2, "title" : "《懺悔錄》", "artist" : "讓・雅克・盧梭（法）", "year" : 1782,"price" : NumberDecimal("280.00") },
{ "_id" : 3, "title" : "《罪與罰》", "artist" : "陀斯妥耶夫斯基（俄）", "year" : 1866,"price" : NumberDecimal("76.04") },
{ "_id" : 4, "title" : "《復活》", "artist" : "列夫・托爾斯泰（俄）","year" : 1899,"price" : NumberDecimal("167.30") },
{ "_id" : 5, "title" : "《九三年》", "artist" : "維克多・雨果（法）", "year" : 1895,"price" : NumberDecimal("483.00") },
{ "_id" : 6, "title" : "《尤利西斯》", "artist" : "詹姆斯・喬伊斯（愛爾蘭）", "year" : 1922,"price" : NumberDecimal("385.00") },
{ "_id" : 7, "title" : "《魔山》", "artist" : "托馬斯・曼（德）", "year" : 1924/* No price*/ },
{ "_id" : 8, "title" : "《永別了，武器》", "artist" : "歐內斯特・海明威（美）", "year" : 1929,"price" : NumberDecimal("118.42") }
])
$bucket 按範圍分組，手動指定各個分組的邊界，用法如下：
db.books.aggregate( [
{
$bucket: {
groupBy: "$price", //用於分組的表示式，這裡使用價格進行分組
boundaries: [ 0, 200, 400 ], //分組邊界，這裡分組邊界為[0,200),[200,400)和其他
default: "Other", //不在[0,200)和[200,400)範圍內的資料放在_id為Other的bucket中
output: {
"count": { $sum: 1 }, //bucket中document的個數
"titles" : { $push: {title:"$title",price:"$price"} } //bucket中的titles
}
}
}
] )
執行上邊的聚合操作，結果如下：
[
{
"_id" : 0,
"count" : 4,
"titles" : [
{"title" : "《白鯨》","price" : NumberDecimal("199.99")},
{"title" : "《罪與罰》","price" : NumberDecimal("76.04")},
{"title" : "《復活》","price" : NumberDecimal("167.30")},
{"title" : "《永別了，武器》","price" : NumberDecimal("118.42")}
]
},
{
"_id" : 200,
"count" : 2,
"titles" : [
{"title" : "《懺悔錄》","price" : NumberDecimal("280.00")},
{"title" : "《尤利西斯》","price" : NumberDecimal("385.00")}
]
},
{
"_id" : "Other",
"count" : 2,
"titles" : [
{"title" : "《九三年》","price" : NumberDecimal("483.00")},
{"title" : "《魔山》"}
]
}
]
View Code
$bucketAuto 和$bucket作用類似，區別在於$bucketAuto不指定分組的邊界，而是指定分組的個數，分組邊界是自動生成的
db.books.aggregate([
{
$bucketAuto: {
groupBy: "$year",
buckets: 3,
output:{
count:{$sum:1},
title:{$push:{title:"$title",year:"$year"}}
}
}
}
])
執行上邊的聚合操作，結果如下：
[
{
"_id" : {
"min" : 1782,
"max" : 1895
},
"count" : 3,
"title" : [
{"title" : "《懺悔錄》","year" : 1782},
{"title" : "《白鯨》","year" : 1845},
{"title" : "《罪與罰》","year" : 1866}
]
},
{
"_id" : {
"min" : 1895,
"max" : 1924
},
"count" : 3,
"title" : [
{"title" : "《九三年》","year" : 1895},
{"title" : "《復活》","year" : 1899},
{"title" : "《尤利西斯》","year" : 1922}
]
},
{
"_id" : {
"min" : 1924,
"max" : 1929
},
"count" : 2,
"title" : [
{"title" : "《魔山》","year" : 1924},
{"title" : "《永別了，武器》","year" : 1929}
]
}
]
View Code
$sample 用於隨機抽取指定數量的document
db.books.aggregate([
{ $sample:{size:2} }
])
執行後隨即抽取了2條document，結果如下，注意因為是隨即抽取的，所以每次執行的結果不同。
[{"_id":6,"title":"《尤利西斯》","artist":"詹姆斯・喬伊斯（愛爾蘭）","year":1922,"price":NumberDecimal("385.00")},
{"_id":8,"title":"《永別了，武器》","artist":"歐內斯特・海明威（美）","year":1929,"price":NumberDecimal("118.42")}]
通過上邊三個栗子，我們應該已經對聚合管道有了一定的理解，其實各種聚合運算子的用法都比較簡單，怎麼靈活的組合各種聚合操作以達到聚合的目的才是我們考慮的重點。
2 mapReduce
MapReduce是一種程式設計模型，用於大規模資料集的並行運算，MapReduce採用"分而治之"的思想，簡單的說就是：將待處理的大資料集分為若干個小資料集，對各個小資料集進行計算獲取中間結果,最後整合中間結果獲取最終的結果。mongoDB也實現了MapReduce，用法還是比較簡單的，語法如下：
db.collection.mapReduce(
function() {emit(key,value);}, //map 函式
function(key,values) {return reduceFunction}, //reduce 函式
{
out: collection, //輸出
query: document, //查詢條件，在Map函式執行前過濾出符合條件的documents
sort: document, //再Map前對documents進行排序
limit: number //傳送給map函式的document條數
}
)
其中：map： 對映函式，遍歷符合query查詢條件的所有document，獲取一系列的鍵值對key-values，如果一個key對應多個value，多個value以陣列形式儲存。
reduce： 統計函式，reduce函式的引數是map函式傳送的key/value集合
out： 指定將統計結果存放在哪個collection中 (不指定則使用臨時集合,在客戶端斷開後自動刪除)。
query：篩選條件，只有滿足條件的文件才會傳送給map函式；
sort：和limit結合使用，在將documents發往map函式前先排序；
limit：和sort結合使用，設定發往map函式的document條數。
我們通過官網的栗子來理解mapReduce的用法，命令如下：
//新增測試資料
db.orders.insertMany([
{cust_id:"A123",amount:500,status:"A"},
{cust_id:"A123",amount:250,status:"A"},
{cust_id:"B212",amount:200,status:"A"},
{cust_id:"A123",amount:300,status:"D"}
])
//mapReduce
db.orders.mapReduce(
function() {emit(this.cust_id,this.amount)}, //map函式，遍歷documents key為cust_id值，values為amount值，或者陣列[amount1,amount2...]
function(key,values){return Array.sum(values)}, //reduce函式，返回合計結果(只會統計values是陣列)
{
query:{status:"A"}, //過濾條件，只向map函式傳送status="A"的documents
out:"myresultColl" //結果存放在myresultColl集合中，如果沒有名字為myresultCOll的集合則新建一個，如果集合存在的話覆蓋舊值
}
)
printjson(db.myresultColl.find() .toArray())
mongoDB的MapReduce可以簡單分為兩個階段：
Map階段：
栗子中的map函式為 function() {emit(this.cust_id,this.amount)} ，執行map函式前先進行query過濾，找到 status=A 的documents，然後將過濾後的documents傳送給map函式，map函式遍歷documents，將cust_id作為key，amount作為value，如果一個cust_id有多個amount值時，value為陣列[amount1,amount2..]，栗子的map函式獲取的key/value集合中有兩個key/value對： {“A123”:[500,250]}和{“B212”:200}
Reduce階段:
reduce函式封裝了我們的聚合邏輯，reduce函式會逐個計算map函式傳過去的key/value對，在上邊栗子中的reduce函式的目的是計算amount的總和。
上邊栗子最終結果存放在集合myresultColl中(out的作用)，通過命令 db.myresultColl.find() 檢視結果如下：
{"_id" : "A123","value" : 750},
{"_id" : "B212","value" : 200}
]
MapReduce屬於輕量級的集合框架，沒有聚合管道那麼多運算子，聚合的思路也比較簡單：先寫一個map函式用於確定需要整合的key/value對(就是我們感興趣的欄位)，然後寫一個reduce函式，其內部封裝著我們的聚合邏輯，reduce函式會逐一處理map函式獲取的key/value對，以此獲取聚合結果。
小結
本節通過幾個栗子簡單介紹了mongoDB的聚合框架：集合管道和MapReduce，聚合管道提供了十分豐富的運算子，讓我們可以方便地進行各種聚合操作，因為聚合管道使用的是mongoDB內建函式所以計算效率一般不會太差。需要注意：①管道處理結果會放在一個document中，所以處理結果不能超過16M(Bson格式的最大尺寸)，②聚合過程中記憶體不能超過100M(可以通過設定{“allowDiskUse”: True}來解決)；MapReduce的map函式和reduce函式都是我們自定義的js函式，這種聚合方式比聚合管道更加靈活，我們可以通過編寫js程式碼來處理複雜的聚合任務，MapReduce的缺點是聚合的邏輯需要我們自己編碼實現。綜上，對於一些簡單的固定的聚集操作推薦使用聚合管道，對於一些複雜的、大量資料集的聚合任務推薦使用MapReduc