Implementando o MapReduce no MongoDB

     Em nosso post Descobrindo o MapReduce, conceituamos e explicamos os elementos fundamentais deste modelo de programação paralelo. No post de hoje, vamos implementar esta função no MongoDB.

Para nosso exemplo, vamos trabalhar com o dataset zips que contém dados sobre as cidades (latitude, longitude, população e CEPs) e estados dos EUA. Ele pode ser baixado neste link: download dataset. Insira os dados utilizando o comando mongoimport (veja neste post como fazer: Como importar arquivos no MongoDB).

     O nosso objeto é contar a quantidade de cidades distintas que possuímos em nosso banco de dados. As cidades se repetem pois temos o CEP como _id de nossos documento, e uma cidade pode possuir diversos CEPs. Vale ressaltar que este exemplo é para exemplificar a implementação da função. Por isso, trabalharemos com apenas um server, sem sharding (veremos em posts futuros como organizar um cluster).

     Para começar, vamos utilizar o comando findOne() para lembrar a estrutura de nossa collection.

db.zips.findOne()

     

     Na sequência, vamos criar duas variáveis que serão utilizadas no MapReduce.

var fonte = db.zips;
var destino = db.distinct;
destino.drop();

     A variável fonte contém a nossa collection. A variável destino contém a collection auxiliar no processo. O comando destino.drop() apaga uma eventual collection que exista com este mesmo nome (procedimento de segurança).

     Agora, vamos escrever a função Map:

map = function() {
  emit( this.city , {count: 1});
}

     Utilizamos a sintaxe this.city pois city é o campo que desejamos realizar a contagem. E, como nossa operação é contar, utilizamos {count:1}.

     Agora, a função Reduce:

reduce = function(key, values) {
  var count = 0;

  values.forEach(function(v) {
    count += v['count'];   
  });

  return {count: count};
};




        Aqui, temos a inicialização de uma variável count = 0 que será incrementada para cada valor que for encontrado durante o loop values.forEach. A saída é nossa contagem, representada pela linha return {count:count}.

     Por fim, consolidamos as duas funções para ter o processo completo de MapReduce:

res = fonte.mapReduce( map, reduce, 
    { out: 'distinct', 
     verbose: true
    }
    );

     Nesta etapa, dizemos ao MongoDB que utilizaremos o mapReduce na collection fonte (fonte.mapReduce) que na verdade é a nossa db.zips, definida como variável no início do processo. Em seguida, passamos como parâmetro as funções criadas map  e reduce, e dizemos que nossa saída será distinta, com o parâmetro out:'distinct'.

     Aqui temos uma observação: o parâmetro verbose:true nos dará a contagem de tempo em casa fase do processo de MapReduce. Ou seja, ele explica qual foi a parcela dos 2360 milissegundos gastos com cada fase do processo.

     A nossa resposta desejada (a contagem de quantas cidades distintas temos em nossa base zips) é dada pelo campo output, com valor de 16.698, de um total de 29.467 documentos de entrada (input). Vemos também que foram reduzidos (reduce) 6.713 documentos, que eram as cidades que tínhamos repetidas na base.

     Se quisermos ter apenas a saída final representada de um modo mais elegante, executamos no shell:

print( "Contagem de cidades distintas= ", destino.count() );

onde contamos a quantidade de documentos na collection auxiliar destino, configurada no início do processo.

     Alternativamente, temos a opção de não criar esta collection exibir a saída com o comando:

print( "distinct count= " + res.counts.output );

     Criamos a collection destino para poder, a qualquer momento, resgatar a quantidade de cidade distintas. Caso tivéssemos apenas a variável res, poderíamos perde-la e ter que executar o processo novamente, o que pode ser muito custoso computacionalmente caso estejamos trabalhando com uma grande quantidade de dados e diversos servidores. Se executarmos o comando show collections encontraremos nossa collection destino armazenada em nosso banco.

     Esperamos ter contribuído com o entendimento desta poderosa ferramenta.

     Até o próximo post!