【mongoDB高級篇②】大數據彙集運算之mapReduce(映射化簡)

簡述

mapReduce從字面上來理解就是兩個過程：map映射以及reduce化簡。是一種比較先進的大數據處理方法，其難度不高，從性能上來講屬於比較暴力的（經過N臺服務器同時來計算），但相較於group以及aggregate來講，功能更強大，並更加靈活。

1. 映射過程:先把某一類數據分組歸類,這裏的映射過程是支持分佈式的,一邊遍歷每一臺服務器,一邊進行分類。

2. 化簡過程:而後再在分組中進行運算，這裏的化簡過程也是支持分佈式的，在分類的過程當中直接運算了。也就是說若是是一個求和的過程，先在a服務器分組求和，而後再在b服務器分組求和····最後再把化簡之後的數據進行最終處理。在映射化簡的過程都是每臺服務器本身的CPU在運算，大量的服務器同時來進行運算工做，這就是大數據基本理念。

在這個映射化簡操做中，MongoDB對每一個輸入文檔（例如集合中知足查詢條件的文檔）執行了map操做。映射操做輸出了鍵值對結果。對那些有多個值的關鍵字，MongoDB執reduce操做，收集並壓縮了最終的聚合結果。而後MongoDB把結果保存到一個集合中。化簡函數還能夠把結果輸出到finalize函數，進一步對聚合結果作處理，固然這步是可選的。

在MongoDB中，全部的映射化簡函數都是使用JavaScript編寫，而且運行在 mongod 進程中。映射化簡操做使用一個集合中文檔做爲輸入，而且能夠在映射階段以前執行任意的排序和限定操做。 mapReduce 命令能夠把結果做爲一個文檔來返回，也能夠把結果寫入集合。輸入集合和輸出集合能夠是分片的。

語法參數

更多參考: http://docs.mongodb.org/manual/reference/command/mapReduce/

map: function() {emit(this.cat_id,this.goods_number); }, # 函數內部要調用內置的emit函數,cat_id表明根據cat_id來進行分組,goods_number表明把文檔中的goods_number字段映射到cat_id分組上的數據,其中this是指向向前的文檔的,這裏的第二個參數能夠是一個對象,若是是一個對象的話,也是做爲數組的元素壓進數組裏面;
 
 reduce: function(cat_id,all_goods_number) {return Array.sum(all_goods_number)}, # cat_id表明着cat_id當前的這一組,all_goods_number表明當前這一組的goods_number集合,這部分返回的就是結果中的value值;
 
 out: <output>, # 輸出到某一個集合中,注意本屬性來還支持若是輸出的集合若是已經存在了,那是替換,合併仍是繼續reduce? 另外還支持輸出到其餘db的分片中,具體用到時查閱文檔,篩選出現的鍵名分別是_id和value;
 
 query: <document>, # 一個查詢表達式,是先查詢出來,再進行mapReduce的
 
 sort: <document>, # 發往map函數前先給文檔排序
 
 limit: <number>, # 發往map函數的文檔數量上限,該參數貌似不能用在分片模式下的mapreduce
 
 finalize: function(key, reducedValue) {return modifiedObject; }, # 從reduce函數中接受的參數key與reducedValue,而且能夠訪問scope中設定的變量
 
 scope: <document>, # 指定一個全局變量,能應用於finalize和reduce函數
 
 jsMode: <boolean>, # 布爾值，是否減小執行過程當中BSON和JS的轉換，默認true,true時BSON-->js-->map-->reduce-->BSON,false時 BSON-->JS-->map-->BSON-->JS-->reduce-->BSON,可處理很是大的mapreduce。
 
 verbose: <boolean> # 是否產生更加詳細的服務器日誌，默認true

實例

簡單應用實例

# 求每組的庫存總量
  var map = function(){
    emit(this.cat_id,this.goods_number);
  }
  var reduce = function(cat_id,numbers){
    return Array.sum(numbers);
  }
  db.goods.mapReduce(map,reduce,{out:'res'})

# 查看Array支持的方法
  for(var i in Array){
    printjson(i);
  }

  "contains"
  "unique"
  "shuffle"
  "tojson"
  "fetchRefs"
  "sum"
  "avg"
  "stdDev"

# 求每一個欄目的平均價格
var map = function(){
  emit(this.cat_id,this.shop_price);
}

var reduce = function(cat_id,prices){
  var avgprice = Array.avg(prices);
  return Math.round(avgprice,2);
}
db.goods.mapReduce(map,reduce,{out:'res'});



# 求出每組的最大價格
var map = function(){
  emit(this.cat_id,this.shop_price);
}

//錯誤操做 ↓↓ 應該在finalize函數中作處理
var reduce = function(cat_id,prices){
  var max = 0;
  for(var i in prices){
    if(i > max) 
      max = i;
  }
  return max;
}

var reduce = function(cat_id,prices){
  return {cat_id:cat_id,prices:prices};
}

var finalize = function(cat_id, prices) {
  var max = 0; 
  if(prices.prices !== null){
    var obj = prices.prices;
    for(var i in obj){
      if(obj[i] > max)
        max = obj[i]
    }
  }
  return max == 0 ? prices : max;
}

db.goods.mapReduce(map,reduce,{out:'res1',finalize:finalize,query:{'shop_price':{$gt:0}}});

# 得到每組的商品集合
var map = function(){
  emit(this.cat_id,this.goods_name);
}

var reduce = function(cat_id,goods_names){
  return {cat_id:cat_id,goods_names:goods_names}
}

var finalize = function(key, reducedValue) {
    return reducedValue == null ? 'none value' : reducedValue; //對reduce的值進行二次處理
}

db.runCommand({
  mapReduce:'goods',
  map:map,
  reduce:reduce,
  finalize:finalize,
  out:'res2'
})


# 對於price大於100的才進行分組映射

## 方法1:

var map = function(){
  if(this.shop_price > 100){
    emit(this.cat_id,{name:this.goods_name,price:this.shop_price});
  }  
}

var reduce = function(cat_id,goods_names){
  return {cat_id:cat_id,goods_names:goods_names}
}

db.runCommand({
  mapReduce:'goods',
  map:map,
  reduce:reduce,
  out:'res2'
})

## 方法2 首推此方法

var map = function(){
  emit(this.cat_id,{name:this.goods_name,price:this.shop_price}); 
}

var reduce = function(cat_id,goods_names){
  return {cat_id:cat_id,goods_names:goods_names}
}

db.runCommand({
  mapReduce:'goods',
  map:map,
  reduce:reduce,
  query:{'shop_price':{$gt:100}},
  out:'res2'
})

官網實例

# 數據結構
{
     _id: ObjectId("50a8240b927d5d8b5891743c"),
     cust_id: "abc123",
     ord_date: new Date("Oct 04, 2012"),
     status: 'A',
     price: 25,
     items: [ { sku: "mmm", qty: 5, price: 2.5 },
              { sku: "nnn", qty: 5, price: 2.5 } ]
}

# 計算每一個顧客的總金額

var mapFunction1 = function() {
   emit(this.cust_id, this.price);
};

var reduceFunction1 = function(keyCustId, valuesPrices) {
  return Array.sum(valuesPrices);
};

db.orders.mapReduce(
 mapFunction1,
 reduceFunction1,
 { out: "map_reduce_example" }
)


# 計算訂單總量和每種 sku 訂購量的平均值
var mapFunction2 = function() {
   for (var idx = 0; idx < this.items.length; idx++) {
       var key = this.items[idx].sku;
       var value = {
                     count: 1,
                     qty: this.items[idx].qty
                   };
       emit(key, value);
   }
};

var reduceFunction2 = function(keySKU, countObjVals) {
     reducedVal = { count: 0, qty: 0 };
     for (var idx = 0; idx < countObjVals.length; idx++) {
         reducedVal.count += countObjVals[idx].count;
         reducedVal.qty += countObjVals[idx].qty;
     }
     return reducedVal;
};

var finalizeFunction2 = function (key, reducedVal) {
   reducedVal.avg = reducedVal.qty/reducedVal.count;
   return reducedVal;
};

db.orders.mapReduce( 
    mapFunction2,
    reduceFunction2,
     {
       out: { merge: "map_reduce_example" },
       query: { ord_date:
                  { $gt: new Date('01/01/2012') }
              },
       finalize: finalizeFunction2
     }
)