Map-Reduce和分片集合

MongoDB支持在分片集合上進行Map-reduce運算,既能夠將分片集合做爲運算的輸入,也能夠做爲輸出。本章節將會講解分片集合上 mapReduce 的注意事項.

將分片集合做爲輸入

當使用分片集合做爲map-reduce的輸入時, mongos 會自動將map-reduce任務分發到全部分片服務器以並行執行，不須要額外的選項, mongos 會等待全部分片返回運算結果。

將分片集合做爲輸出

若是 mapReduce 命令的 out 字段有 sharded 值，MongoDB會用 _id 作片鍵將輸出進行分片.

將結果輸出到分片集合中

• 若是輸出的集合不存在，MongoDB會建立並使用 _id 做爲片鍵.

• 對於新的或者空的分片集合，MongoDB使用map-reduce最初階段產生的結果進行 數據塊 的初始化。

• mongos 併發的爲每一個擁有數據塊的分片分發map-reduce的後處理命令.在後處理階段，每一個分片服務器會從其餘分片讀取屬於本身分塊的數據，執行最終的reduce/finalize，並把結果寫到本地的輸出集合。

  在 mongo 命令行， db.collection.mapReduce() 方法封裝了 mapReduce 命令。下面是一些使用 db.collection.mapReduce() 方法的例子：

  接下來的map-reduce操做都是在集合 orders 上執行，集合中的文檔格式相似於：

  {
       _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 } ]
  }

  計算每一個顧客的總金額

  首先在 orders 集合上按 cust_id 字段值執行分組map-reduce操做，並對每一個分組內文檔的 price 字段進行求和操做。

• 定義map方法來處理每個輸入文檔：

  • 在方法中，this 指的是當前 map-reduce 操做正在處理的文檔。

  • 該方法把每個文檔的 price 和 cust_id 字段映射爲一對，並提交 cust_id 和 price 的配對。

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

• 定義對應的reduce函數，入參是 keyCustId 和 valuesPrices:

  • valuesPrices 字段是一個數組，保存了由map函數提交的按 keyCustId 分組的多個 price 值。

  • reduce函數最終對 valuesPrice 數組內的元素值執行求和運算。

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

• 使用 mapFunction1 方法和 reduceFunction1 方法對 orders 集合中的文檔執行 map-reduce。

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

  本次操做的結果輸出到 map_reduce_example 集合中。若是 map_reduce_example 集合已經存在，本次操做會把舊的記錄覆蓋。

  計算訂單總量和每種 sku 訂購量的平均值

  在這個例子中，會對集合 orders 中全部的 ord_date 大於 01/01/2012 的文檔執行map-reduce操做。該操做對全部文檔按 item.sku 字段的值進行分組，並計算訂單總數和每種 sku 訂購量的總和，同時也會計算每種 sku 的平均值。

• 定義map方法來處理每個輸入文檔：

  • 在方法中，this 指的是當前 map-reduce 操做正在處理的文檔。

  • 該方法逐個處理文檔中的每一個名目，爲每一個名目建立一個 sku 和 value 的聯合，

  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);
                         }
                      };

• 定義相應的reduce函數，它使用兩個參數 keySKU 和 countObjVals：

  • countObjVals 是一個數組字段，保存了從map函數提交給reduce函數的分組後的多個 keySKU 值。

  • 該方法對 countObjVals 數組進行reduce，轉換爲一個單獨的對象 reducedValue

  • 在 reducedVal 中， 字段 count 的值是對數組中每一個元素中的 count 值求和的結果，qty 字段的值是對對數組中每一個元素中的 qty 值求和的結果。

  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;
                    };

• 定義一個使用兩個參數 key 和 reducedVal 的結束函數。該函數在 reducedVal 中添加一個平均值 avg 字段，而後返回修改後的對象：

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

• 在 orders 集合上執行使用了 mapFunction2, reduceFunction2， 和 finalizeFunction2 方法的 map-reduce 操做。

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

  This operation uses the query field to select only those documents with o