https://segmentfault.com/a/1190000004263358

【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,);
}
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 = ;
  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 = ; 
  if(prices.prices !== null){
    var obj = prices.prices;
    for(var i in obj){
      if(obj[i] > max)
        max = obj[i]
    }
  }
  return max ==  ? prices : max;
}

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

# 獲得每組的商品集合
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 > ){
    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:}},
  out:'res2'
})                

官網執行個體

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

# 計算每個顧客的總金額

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 = ; idx < this.items.length; idx++) {
       var key = this.items[idx].sku;
       var value = {
                     count: ,
                     qty: this.items[idx].qty
                   };
       emit(key, value);
   }
};

var reduceFunction2 = function(keySKU, countObjVals) {
     reducedVal = { count: , qty:  };
     for (var idx = ; 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
     }
)                

• 2016年01月16日釋出

• 新浪微網誌
• 微信
• Twitter
• Facebook