Menu Docs
Página inicial do Docs
/
Manual do MongoDB
/
Operações de agregação
/
Map-Reduce

Map-reduce para aggregation pipeline

Um pipeline de agregação fornece melhor desempenho e usabilidade do que uma operaçãode redução de mapa.

As operações de redução de mapa podem ser reescritas usando pipeline de agregação, como $group e $merge.

Para operações de map-reduce que exigem funcionalidade personalizada, o MongoDB fornece os operadores de agregação $accumulator e $function. Use esses operadores para definir expressões de agregação personalizadas no JavaScript.

Expressões de redução de mapa podem ser reescritas como mostrado nas seções seguintes.

Tabela de tradução do Map-Reduce to pipeline de agregação

A tabela é apenas uma tradução aproximada. Por exemplo, a tabela mostra uma tradução aproximada do mapFunction utilizando o $project.

  • No entanto, a lógica do mapFunction pode exigir estágios adicionais, como se a lógica incluir iteração em uma array:

    function() {
       this.items.forEach(function(item){ emit(item.sku, 1); });
    }

    Em seguida, o pipeline de agregação inclui um $unwind e um $project:

    { $unwind: "$items "},
    { $project: { emits: { key: { "$items.sku" }, value: 1 } } },

  • O campo emits em $project pode ter outro nome. Para comparação visual, o nome do campo emits foi escolhido.

Map-Reduce
Pipeline de agregação
db.collection.mapReduce(
<mapFunction>,
<reduceFunction>,
{
query: <queryFilter>,
sort: <sortOrder>,
limit: <number>,
finalize: <finalizeFunction>,
out: <collection>
}
)
db.collection.aggregate( [
{ $match: <queryFilter> },
{ $sort: <sortOrder> },
{ $limit: <number> },
{ $project: { emits: { k: <expression>, v: <expression> } } },
{ $unwind: "$emits" },
{ $group: {
_id: "$emits.k"},
value: { $accumulator: {
init: <initCode>,
accumulate: <reduceFunction>,
accumulateArgs: [ "$emit.v"],
merge: <reduceFunction>,
finalize: <finalizeFunction>,
lang: "js" }}
} },
{ $out: <collection> }
] )
db.collection.mapReduce(
<mapFunction>,
<reduceFunction>,
{
query: <queryFilter>,
sort: <sortOrder>,
limit: <number>,
finalize: <finalizeFunction>,
out: { replace: <collection>, db:<db> }
}
)
db.collection.aggregate( [
{ $match: <queryFilter> },
{ $sort: <sortOrder> },
{ $limit: <number> },
{ $project: { emits: { k: <expression>, v: <expression> } } },
{ $unwind: "$emits" },
{ $group: {
_id: "$emits.k"},
value: { $accumulator: {
init: <initCode>,
accumulate: <reduceFunction>,
accumulateArgs: [ "$emit.v"],
merge: <reduceFunction>,
finalize: <finalizeFunction>,
lang: "js" }}
} },
{ $out: { db: <db>, coll: <collection> } }
] )
db.collection.mapReduce(
<mapFunction>,
<reduceFunction>,
{
query: <queryFilter>,
sort: <sortOrder>,
limit: <number>,
finalize: <finalizeFunction>,
out: { merge: <collection>, db: <db> }
}
)
db.collection.aggregate( [
{ $match: <queryFilter> },
{ $sort: <sortOrder> },
{ $limit: <number> },
{ $project: { emits: { k: <expression>, v: <expression> } } },
{ $unwind: "$emits" },
{ $group: {
_id: "$emits.k"},
value: { $accumulator: {
init: <initCode>,
accumulate: <reduceFunction>,
accumulateArgs: [ "$emit.v"],
merge: <reduceFunction>,
finalize: <finalizeFunction>,
lang: "js" }}
} },
{ $merge: {
into: { db: <db>, coll: <collection>},
on: "_id"
whenMatched: "replace",
whenNotMatched: "insert"
} },
] )
db.collection.mapReduce(
<mapFunction>,
<reduceFunction>,
{
query: <queryFilter>,
sort: <sortOrder>,
limit: <number>,
finalize: <finalizeFunction>,
out: { reduce: <collection>, db: <db> }
}
)
db.collection.aggregate( [
{ $match: <queryFilter> },
{ $sort: <sortOrder> },
{ $limit: <number> },
{ $project: { emits: { k: <expression>, v: <expression> } } },
{ $unwind: "$emits" },
{ $group: {
_id: "$emits.k"},
value: { $accumulator: {
init: <initCode>,
accumulate: <reduceFunction>,
accumulateArgs: [ "$emit.v"],
merge: <reduceFunction>,
finalize: <finalizeFunction>,
lang: "js" }}
} },
{ $merge: {
into: { db: <db>, coll: <collection> },
on: "_id"
whenMatched: [
{ $project: {
value: { $function: {
body: <reduceFunction>,
args: [
"$_id",
[ "$value", "$$new.value" ]
],
lang: "js"
} }
} }
]
whenNotMatched: "insert"
} },
] )
db.collection.mapReduce(
<mapFunction>,
<reduceFunction>,
{
query: <queryFilter>,
sort: <sortOrder>,
limit: <number>,
finalize: <finalizeFunction>,
out: { inline: 1 }
}
)
db.collection.aggregate( [
{ $match: <queryFilter> },
{ $sort: <sortOrder> },
{ $limit: <number> },
{ $project: { emits: { k: <expression>, v: <expression> } } },
{ $unwind: "$emits" },
{ $group: {
_id: "$emits.k"},
value: { $accumulator: {
init: <initCode>,
accumulate: <reduceFunction>,
accumulateArgs: [ "$emit.v"],
merge: <reduceFunction>,
finalize: <finalizeFunction>,
lang: "js" }}
} }
] )

Exemplos

Várias expressões de redução de mapa podem ser reescritas utilizando operadores de pipeline de agregação, como $group, $merge e outros, sem exigir funções personalizadas. No entanto, para fins ilustrativos, os exemplos a seguir fornecem ambas as alternativas.

Exemplo 1

A seguinte operação de redução de mapa nos grupos de coleta do orders pelo cust_id e calcula a soma do price para cada cust_id:

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

Alternativa 1: (Recomendado) Você pode reescrever a operação em um pipeline de agregação sem traduzir a função map-reduce para estágios de pipeline equivalentes:

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

Alternativa 2: (Apenas para fins ilustrativos) O pipeline de agregação fornece uma tradução das várias funções de redução de mapa, usando $accumulator para definir funções personalizadas:

db.orders.aggregate( [
   { $project: { emit: { key: "$cust_id", value: "$price" } } },  // equivalent to the map function
   { $group: {                                                    // equivalent to the reduce function
         _id: "$emit.key",
         valuesPrices: { $accumulator: {
                     init: function() { return 0; },
                     initArgs: [],
                     accumulate: function(state, value) { return state + value; },
                     accumulateArgs: [ "$emit.value" ],
                     merge: function(state1, state2) { return state1 + state2; },
                     lang: "js"
         } }
   } },
   { $out: "agg_alternative_2" }
] )

  1. Primeiro, o estágio $project gera documentos com um campo emit. O campo emit é um documento com os campos:

    • key que contém o valor cust_id para o documento

    • value que contém o valor price para o documento

    { "_id" : 1, "emit" : { "key" : "Ant O. Knee", "value" : 25 } }
    { "_id" : 2, "emit" : { "key" : "Ant O. Knee", "value" : 70 } }
    { "_id" : 3, "emit" : { "key" : "Busby Bee", "value" : 50 } }
    { "_id" : 4, "emit" : { "key" : "Busby Bee", "value" : 25 } }
    { "_id" : 5, "emit" : { "key" : "Busby Bee", "value" : 50 } }
    { "_id" : 6, "emit" : { "key" : "Cam Elot", "value" : 35 } }
    { "_id" : 7, "emit" : { "key" : "Cam Elot", "value" : 25 } }
    { "_id" : 8, "emit" : { "key" : "Don Quis", "value" : 75 } }
    { "_id" : 9, "emit" : { "key" : "Don Quis", "value" : 55 } }
    { "_id" : 10, "emit" : { "key" : "Don Quis", "value" : 25 } }

  2. Em seguida, o $group utiliza o operador $accumulator para adicionar os valores emitidos:

    { "_id" : "Don Quis", "valuesPrices" : 155 }
    { "_id" : "Cam Elot", "valuesPrices" : 60 }
    { "_id" : "Ant O. Knee", "valuesPrices" : 95 }
    { "_id" : "Busby Bee", "valuesPrices" : 125 }

  3. Finalmente, o $out grava a saída na coleção agg_alternative_2. Alternativamente, você pode utilizar $merge ao invés de $out.

Exemplo 2

A seguinte operação de redução de mapa nos grupos de coleta do orders pelo campo item.sku e calcula o número de pedidos e a quantidade total solicitada para cada SKU. Em seguida, a operação calcula a quantidade média por pedido para cada valor de sku e mescla os resultados na coleção de saída.

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_example2" },
      query: { ord_date: { $gte: new Date("2020-03-01") } },
      finalize: finalizeFunction2
   }
   );

Alternativa 1: (Recomendado) Você pode reescrever a operação em um pipeline de agregação sem traduzir a função map-reduce para estágios de pipeline equivalentes:

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

Alternativa 2: (Apenas para fins ilustrativos) O pipeline de agregação fornece uma tradução das várias funções de redução de mapa, usando $accumulator para definir funções personalizadas:

db.orders.aggregate( [
      { $match: { ord_date: {$gte: new Date("2020-03-01") } } },
      { $unwind: "$items" },
      { $project: { emit: { key: "$items.sku", value: { count: { $literal: 1 }, qty: "$items.qty" } } } },
      { $group: {
            _id: "$emit.key",
            value: { $accumulator: {
               init: function() { return { count: 0, qty: 0 }; },
               initArgs: [],
               accumulate: function(state, value) {
                  state.count += value.count;
                  state.qty += value.qty;
                  return state;
               },
               accumulateArgs: [ "$emit.value" ],
               merge: function(state1, state2) {
                  return { count: state1.count + state2.count, qty: state1.qty + state2.qty };
               },
               finalize: function(state) {
                  state.avg = state.qty / state.count;
                  return state;
               },
               lang: "js"}
            }
      } },
      { $merge: {
         into: "agg_alternative_4",
         on: "_id",
         whenMatched: "replace",
         whenNotMatched: "insert"
      } }
] )

  1. A etapa $match seleciona apenas os documentos com ord_date maior ou igual a new Date("2020-03-01").

  2. O estágio $unwind divide o documento pelo campo de array items para gerar um documento para cada elemento da array. Por exemplo:

    { "_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" }
    { "_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" }
    { "_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" }
    { "_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" }
    { "_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" }
    { "_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" }
    { "_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" }
    { "_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" }
    ...

  3. O estágio $project gera documentos com um campo emit. O campo emit é um documento com os campos:

    • key que contém o valor items.sku

    • value que contém um documento com o valor qty e um valor count

    { "_id" : 1, "emit" : { "key" : "oranges", "value" : { "count" : 1, "qty" : 5 } } }
    { "_id" : 1, "emit" : { "key" : "apples", "value" : { "count" : 1, "qty" : 5 } } }
    { "_id" : 2, "emit" : { "key" : "oranges", "value" : { "count" : 1, "qty" : 8 } } }
    { "_id" : 2, "emit" : { "key" : "chocolates", "value" : { "count" : 1, "qty" : 5 } } }
    { "_id" : 3, "emit" : { "key" : "oranges", "value" : { "count" : 1, "qty" : 10 } } }
    { "_id" : 3, "emit" : { "key" : "pears", "value" : { "count" : 1, "qty" : 10 } } }
    { "_id" : 4, "emit" : { "key" : "oranges", "value" : { "count" : 1, "qty" : 10 } } }
    { "_id" : 5, "emit" : { "key" : "chocolates", "value" : { "count" : 1, "qty" : 5 } } }
    ...

  4. O $group usa o operador $accumulator para adicionar os count e qty emitidos e calcular o campo avg:

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

  5. Finalmente, o $merge grava a saída na coleção agg_alternative_4. Se um documento existente tiver a mesma chave _id que o novo resultado, a operação substituirá o documento existente. Se não houver nenhum documento existente com a mesma chave, a operação inserirá o documento.

Dica

Veja também:

Comparação de comandos de agregação

Voltar

Solucionar problemas de redução

Próximo

Indexes