Map-Reduce 例子

    在本页面

    在mongo shell 中，db.collection.mapReduce()方法是MapReduce命令周围的 wrapper。以下示例使用db.collection.mapReduce()方法：

    聚合管道作为替代

    聚合管道 比map-reduce提供更好的性能和更一致的接口。

    各种map-reduce表达式可以使用被重写聚合管道运算符，诸如$group， $merge等

    下面的示例包括聚合管道备选方案。

    orders使用以下文档创建样本集合：

    1. db.orders.insertMany([
    2.    { _id: 1, cust_id: "Ant O. Knee", ord_date: new Date("2020-03-01"), price: 25, items: [ { sku: "oranges", qty: 5, price: 2.5 }, { sku: "apples", qty: 5, price: 2.5 } ], status: "A" },
    3.    { _id: 2, cust_id: "Ant O. Knee", ord_date: new Date("2020-03-08"), price: 70, items: [ { sku: "oranges", qty: 8, price: 2.5 }, { sku: "chocolates", qty: 5, price: 10 } ], status: "A" },
    4.    { _id: 3, cust_id: "Busby Bee", ord_date: new Date("2020-03-08"), price: 50, items: [ { sku: "oranges", qty: 10, price: 2.5 }, { sku: "pears", qty: 10, price: 2.5 } ], status: "A" },
    5.    { _id: 4, cust_id: "Busby Bee", ord_date: new Date("2020-03-18"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
    6.    { _id: 5, cust_id: "Busby Bee", ord_date: new Date("2020-03-19"), price: 50, items: [ { sku: "chocolates", qty: 5, price: 10 } ], status: "A"},
    7.    { _id: 6, cust_id: "Cam Elot", ord_date: new Date("2020-03-19"), price: 35, items: [ { sku: "carrots", qty: 10, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "A" },
    8.    { _id: 7, cust_id: "Cam Elot", ord_date: new Date("2020-03-20"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
    9.    { _id: 8, cust_id: "Don Quis", ord_date: new Date("2020-03-20"), price: 75, items: [ { sku: "chocolates", qty: 5, price: 10 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "A" },
    10.    { _id: 9, cust_id: "Don Quis", ord_date: new Date("2020-03-20"), price: 55, items: [ { sku: "carrots", qty: 5, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 }, { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
    11.    { _id: 10, cust_id: "Don Quis", ord_date: new Date("2020-03-23"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" }
    12. ])

    返回每位客户的总价格

    对orders集合执行map-reduce操作，以对进行分组cust_id，并计算price每个的 的总和cust_id：

    1. 定义map函数来处理每个输入文档：
    • 在函数中，this指的是map-reduce操作正在处理的文档。
    • 该函数将映射price到cust_id每个文档的，并发出cust_id和price对。
    1. var mapFunction1 = function() {
    2.    emit(this.cust_id, this.price);
    3. };
    1. 使用两个参数keyCustId和定义相应的reduce函数 valuesPrices：
    • valuesPrices是一个数组，其元素是price 由map功能发射并由分组值keyCustId。
    • 该函数将valuesPrice数组简化为其元素的总和。
    1. var reduceFunction1 = function(keyCustId, valuesPrices) {
    2.    return Array.sum(valuesPrices);
    3. };
    1. orders使用mapFunction1map函数和reduceFunction1 reduce函数对集合中的所有文档执行map-reduce 。
    1. db.orders.mapReduce(
    2.    mapFunction1,
    3.    reduceFunction1,
    4.    { out: "map_reduce_example" }
    5. )

    此操作将结果输出到名为的集合 map_reduce_example。如果map_reduce_example集合已经存在，则该操作将用此map-reduce操作的结果替换内容。

    1. 查询map_reduce_example集合以验证结果：
    1. db.map_reduce_example.find().sort( { _id: 1 } )

    ​ 该操作返回以下文档：

    1. { "_id" : "Ant O. Knee", "value" : 95 }
    2. { "_id" : "Busby Bee", "value" : 125 }
    3. { "_id" : "Cam Elot", "value" : 60 }
    4. { "_id" : "Don Quis", "value" : 155 }

    聚合替代

    使用可用的聚合管道运算符，您可以重写map-reduce操作，而无需定义自定义函数：

    1. db.orders.aggregate([
    2.    { $group: { _id: "$cust_id", value: { $sum: "$price" } } },
    3.    { $out: "agg_alternative_1" }
    4. ])

    1. $group由平台组cust_id并计算value字段（参见$sum）。该 value字段包含price每个的总计cust_id。

      该阶段将以下文档输出到下一阶段：

      1. { "_id" : "Don Quis", "value" : 155 }
      2. { "_id" : "Ant O. Knee", "value" : 95 }
      3. { "_id" : "Cam Elot", "value" : 60 }
      4. { "_id" : "Busby Bee", "value" : 125 }

    2. 然后，$out将输出写入collection agg_alternative_1。或者，您可以使用 $merge代替$out。

    3. 查询agg_alternative_1集合以验证结果：

      1. db.agg_alternative_1.find().sort( { _id: 1 } )

      该操作返回以下文档：

      1. { "_id" : "Ant O. Knee", "value" : 95 }
      2. { "_id" : "Busby Bee", "value" : 125 }
      3. { "_id" : "Cam Elot", "value" : 60 }
      4. { "_id" : "Don Quis", "value" : 155 }

      用每个项目的平均数量计算订单和总数量

    在此示例中，您将对值大于或等于的orders所有文档在集合上执行map-reduce操作 。工序按字段分组 ，并计算每个的订单数量和总订购量。然后，该操作将为每个值计算每个订单的平均数量，并将结果合并到输出集合中。合并结果时，如果现有文档的密钥与新结果相同，则该操作将覆盖现有文档。如果不存在具有相同密钥的文档，则该操作将插入该文档。

    1. 定义map函数来处理每个输入文档：

      • 在函数中，this指的是map-reduce操作正在处理的文档。
      • 对于每个商品，该函数将其sku与一个新对象相关联，该对象value包含订单的countof 1和该商品qty，并发出skuand value对。
      1. var mapFunction2 = function() {
      2.     for (var idx = 0; idx < this.items.length; idx++) {
      3.        var key = this.items[idx].sku;
      4.        var value = { count: 1, qty: this.items[idx].qty };
      6.        emit(key, value);
      7.     }
      8. };

    2. 使用两个参数keySKU和定义相应的reduce函数 countObjVals：

      • countObjVals是一个数组，其元素是映射到keySKU由map函数传递给reducer函数的分组值的对象。
      • 该函数将countObjVals数组简化为reducedValue包含count和 qty字段的单个对象。
      • 在中reducedVal，该count字段包含 count各个数组元素的qty字段总和，而该字段包含各个数组元素的 字段总和qty。
      1. var reduceFunction2 = function(keySKU, countObjVals) {
      2.    reducedVal = { count: 0, qty: 0 };
      4.    for (var idx = 0; idx < countObjVals.length; idx++) {
      5.        reducedVal.count += countObjVals[idx].count;
      6.        reducedVal.qty += countObjVals[idx].qty;
      7.    }
      9.    return reducedVal;
      10. };

    3. 定义有两个参数的函数确定key和 reducedVal。该函数修改reducedVal对象以添加一个名为avg的计算字段，并返回修改后的对象：

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

    4. 在执行的map-reduce操作orders使用集合mapFunction2，reduceFunction2和 finalizeFunction2功能。

      1. db.orders.mapReduce(
      2.    mapFunction2,
      3.    reduceFunction2,
      4.    {
      5.      out: { merge: "map_reduce_example2" },
      6.      query: { ord_date: { $gte: new Date("2020-03-01") } },
      7.      finalize: finalizeFunction2
      8.    }
      9.  );

      此操作使用该query字段选择仅ord_date大于或等于的那些文档。然后将结果输出到集合 。new Date("2020-03-01") map_reduce_example2

      如果map_reduce_example2集合已经存在，则该操作会将现有内容与此map-reduce操作的结果合并。也就是说，如果现有文档具有与新结果相同的密钥，则该操作将覆盖现有文档。如果不存在具有相同密钥的文档，则该操作将插入该文档。

    5. 查询map_reduce_example2集合以验证结果：

      1. db.map_reduce_example2.find().sort( { _id: 1 } )

      该操作返回以下文档：

      1. { "_id" : "apples", "value" : { "count" : 3, "qty" : 30, "avg" : 10 } }
      2. { "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }
      3. { "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
      4. { "_id" : "oranges", "value" : { "count" : 6, "qty" : 58, "avg" : 9.666666666666666 } }
      5. { "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }

      聚合替代

      使用可用的聚合管道运算符，您可以重写map-reduce操作，而无需定义自定义函数：

      1. db.orders.aggregate( [
      2.    { $match: { ord_date: { $gte: new Date("2020-03-01") } } },
      3.    { $unwind: "$items" },
      4.    { $group: { _id: "$items.sku", qty: { $sum: "$items.qty" }, orders_ids: { $addToSet: "$_id" } }  },
      5.    { $project: { value: { count: { $size: "$orders_ids" }, qty: "$qty", avg: { $divide: [ "$qty", { $size: "$orders_ids" } ] } } } },
      6.    { $merge: { into: "agg_alternative_3", on: "_id", whenMatched: "replace",  whenNotMatched: "insert" } }
      7. ] )

    6. 该$match阶段仅选择ord_date大于或等于new Date("2020-03-01")的那些文档。

    7. 该$unwinds阶段按items数组字段细分文档，以输出每个数组元素的文档。例如：

      1.    { "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 5, "price" : 2.5 }, "status" : "A" }
      2.    { "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "apples", "qty" : 5, "price" : 2.5 }, "status" : "A" }
      3.    { "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "oranges", "qty" : 8, "price" : 2.5 }, "status" : "A" }
      4.    { "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
      5.    { "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
      6.    { "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "pears", "qty" : 10, "price" : 2.5 }, "status" : "A" }
      7.    { "_id" : 4, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-18T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
      8.    { "_id" : 5, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-19T00:00:00Z"), "price" : 50, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
      9.    ...

    8. $group由平台组items.sku，计算每个SKU：

      • 该qty字段。该qty字段包含qty每个订单的总数items.sku（请参阅参考资料$sum）。

      • orders_ids列表。该orders_ids字段包含不同顺序的列表_id的对items.sku（参见 $addToSet）。

        1. { "_id" : "chocolates", "qty" : 15, "orders_ids" : [ 2, 5, 8 ] }
        2. { "_id" : "oranges", "qty" : 63, "orders_ids" : [ 4, 7, 3, 2, 9, 1, 10 ] }
        3. { "_id" : "carrots", "qty" : 15, "orders_ids" : [ 6, 9 ] }
        4. { "_id" : "apples", "qty" : 35, "orders_ids" : [ 9, 8, 1, 6 ] }
        5. { "_id" : "pears", "qty" : 10, "orders_ids" : [ 3 ] }

    9. 该$project阶段调整输出文档的形状以反映map-reduce的输出，该输出具有两个字段_id和 value。该$project设置：

      • value.count到的尺寸orders_ids数组。（请参阅$size）
      • 在value.qty到qty输入文档的数量字段。

      • value.avg平均每笔订购的数量。（请参阅$divide和$size）

        1. { "_id" : "apples", "value" : { "count" : 4, "qty" : 35, "avg" : 8.75 } }
        2. { "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }
        3. { "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
        4. { "_id" : "oranges", "value" : { "count" : 7, "qty" : 63, "avg" : 9 } }
        5. { "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }

    10. 最后，$merge将输出写入collection agg_alternative_3。如果现有文档的密钥_id与新结果相同，则该操作将覆盖现有文档。如果不存在具有相同密钥的文档，则该操作将插入该文档。

    11. 查询agg_alternative_3集合以验证结果：

      1.  db.agg_alternative_3.find().sort( { _id: 1 } )

      该操作返回以下文档：

      1.  { "_id" : "apples", "value" : { "count" : 4, "qty" : 35, "avg" : 8.75 } }
      2.  { "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }
      3.  { "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
      4.  { "_id" : "oranges", "value" : { "count" : 7, "qty" : 63, "avg" : 9 } }
      5.  { "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }

    译者：李冠飞

    校对：

    参见

    原文 - Map-Reduce Examples