examples - MongoDB: rendimiento terrible de MapReduce

mongodb mapreduce ejemplos

Question

Tengo una larga historia con bases de datos relacionales, pero soy nuevo en MongoDB y MapReduce, así que estoy casi seguro de que debo estar haciendo algo mal. Iré directo a la pregunta. Lo siento si es largo.

Tengo una tabla de base de datos en MySQL que rastrea el número de vistas de perfil de los miembros para cada día. Para la prueba tiene 10,000,000 filas.

CREATE TABLE `profile_views` (
  `id` int(10) unsigned NOT NULL auto_increment,
  `username` varchar(20) NOT NULL,
  `day` date NOT NULL,
  `views` int(10) unsigned default ''0'',
  PRIMARY KEY  (`id`),
  UNIQUE KEY `username` (`username`,`day`),
  KEY `day` (`day`)
) ENGINE=InnoDB;

Los datos típicos pueden verse así.

+--------+----------+------------+------+
| id     | username | day        | hits |
+--------+----------+------------+------+
| 650001 | Joe      | 2010-07-10 |    1 |
| 650002 | Jane     | 2010-07-10 |    2 |
| 650003 | Jack     | 2010-07-10 |    3 |
| 650004 | Jerry    | 2010-07-10 |    4 |
+--------+----------+------------+------+

Utilizo esta consulta para obtener los 5 perfiles más vistos desde 2010-07-16.

SELECT username, SUM(hits)
FROM profile_views
WHERE day > ''2010-07-16''
GROUP BY username
ORDER BY hits DESC
LIMIT 5/G

Esta consulta se completa en menos de un minuto. ¡No está mal!

Ahora avanzando hacia el mundo de MongoDB. Configuré un entorno fragmentado usando 3 servidores. Servidores M, S1 y S2. Usé los siguientes comandos para configurar el equipo (Nota: he oscurecido los dispositivos adicionales de IP).

S1 => 127.20.90.1
./mongod --fork --shardsvr --port 10000 --dbpath=/data/db --logpath=/data/log

S2 => 127.20.90.7
./mongod --fork --shardsvr --port 10000 --dbpath=/data/db --logpath=/data/log

M => 127.20.4.1
./mongod --fork --configsvr --dbpath=/data/db --logpath=/data/log
./mongos --fork --configdb 127.20.4.1 --chunkSize 1 --logpath=/data/slog

Una vez que se pusieron en funcionamiento, salté al servidor M y lancé mongo. Emití los siguientes comandos:

use admin
db.runCommand( { addshard : "127.20.90.1:10000", name: "M1" } );
db.runCommand( { addshard : "127.20.90.7:10000", name: "M2" } );
db.runCommand( { enablesharding : "profiles" } );
db.runCommand( { shardcollection : "profiles.views", key : {day : 1} } );
use profiles
db.views.ensureIndex({ hits: -1 });

Luego importé las mismas 10,000,000 filas de MySQL, lo que me dio documentos que se ven así:

{
    "_id" : ObjectId("4cb8fc285582125055295600"),
    "username" : "Joe",
    "day" : "Fri May 21 2010 00:00:00 GMT-0400 (EDT)",
    "hits" : 16
}

Ahora viene la verdadera carne y las patatas aquí ... Mi mapa y reducir las funciones. De vuelta en el servidor M en el shell configuro la consulta y la ejecuto así.

use profiles;
var start = new Date(2010, 7, 16);
var map = function() {
    emit(this.username, this.hits);
}
var reduce = function(key, values) {
    var sum = 0;
    for(var i in values) sum += values[i];
    return sum;
}
res = db.views.mapReduce(
    map,
    reduce,
    {
        query : { day: { $gt: start }}
    }
);

Y he aquí que tuve problemas. ¡Esta consulta tomó más de 15 minutos en completarse! La consulta de MySQL tomó menos de un minuto. Aquí está el resultado:

{
        "result" : "tmp.mr.mapreduce_1287207199_6",
        "shardCounts" : {
                "127.20.90.7:10000" : {
                        "input" : 4917653,
                        "emit" : 4917653,
                        "output" : 1105648
                },
                "127.20.90.1:10000" : {
                        "input" : 5082347,
                        "emit" : 5082347,
                        "output" : 1150547
                }
        },
        "counts" : {
                "emit" : NumberLong(10000000),
                "input" : NumberLong(10000000),
                "output" : NumberLong(2256195)
        },
        "ok" : 1,
        "timeMillis" : 811207,
        "timing" : {
                "shards" : 651467,
                "final" : 159740
        },
}

No solo tardó en ejecutarse, sino que los resultados ni siquiera parecen correctos.

db[res.result].find().sort({ hits: -1 }).limit(5);
{ "_id" : "Joe", "value" : 128 }
{ "_id" : "Jane", "value" : 2 }
{ "_id" : "Jerry", "value" : 2 }
{ "_id" : "Jack", "value" : 2 }
{ "_id" : "Jessy", "value" : 3 }

Sé que esos valores deben ser mucho más altos.

Mi comprensión de todo el paradigma de MapReduce es que la tarea de realizar esta consulta debe dividirse entre todos los miembros del shard, lo que debería aumentar el rendimiento. Esperé a que Mongo terminara de distribuir los documentos entre los dos servidores de fragmentos después de la importación. Cada uno tenía casi exactamente 5,000,000 de documentos cuando comencé esta consulta.

Entonces, debo estar haciendo algo mal. ¿Alguien puede darme algún consejo?

Editar: Alguien en el IRC mencionó agregar un índice en el campo del día, pero hasta donde sé, MongoDB lo hizo automáticamente.

Answer 1

¿Ya has intentado usar el conector hadoop para mongodb?

Mire este enlace aquí: http://docs.mongodb.org/ecosystem/tutorial/getting-started-with-hadoop/

Ya que está utilizando solo 3 fragmentos, no sé si este enfoque mejoraría su caso.

Answer 2

No estás haciendo nada malo. (Además de ordenar el valor incorrecto como ya has notado en tus comentarios).

MongoDB map / reduce el rendimiento simplemente no es tan bueno. Este es un problema conocido; ver por ejemplo http://jira.mongodb.org/browse/SERVER-1197 donde un enfoque ingenuo es ~ 350x más rápido que M / R.

Sin embargo, una ventaja es que puede especificar un nombre de colección de salida permanente con el argumento out de la llamada mapReduce . Una vez que se completa la M / R, la colección temporal cambiará de nombre atómico al nombre permanente. De esta forma, puede programar sus actualizaciones de estadísticas y consultar la colección de resultados de M / R en tiempo real.

Answer 3

Tal vez sea demasiado tarde, pero ...

En primer lugar, está consultando la colección para llenar el MapReduce sin un índice. Debes crear un índice en "día".

MongoDB MapReduce tiene un único hilo en un único servidor, pero se duplica en fragmentos. Los datos en mongo shards se mantienen juntos en fragmentos contiguos ordenados por clave de fragmentación.

Como su clave de fragmentación es "día", y usted la está consultando, probablemente solo esté utilizando uno de sus tres servidores. La clave de Sharding solo se usa para difundir los datos. Map Reduce consultará usando el índice "día" en cada fragmento, y será muy rápido.

Agregue algo delante de la tecla de día para difundir los datos. El nombre de usuario puede ser una buena opción.

De esta forma, la reducción del mapa se lanzará en todos los servidores y, con suerte, reducirá el tiempo en tres.

Algo como esto:

use admin
db.runCommand( { addshard : "127.20.90.1:10000", name: "M1" } );
db.runCommand( { addshard : "127.20.90.7:10000", name: "M2" } );
db.runCommand( { enablesharding : "profiles" } );
db.runCommand( { shardcollection : "profiles.views", key : {username : 1,day: 1} } );
use profiles
db.views.ensureIndex({ hits: -1 });
db.views.ensureIndex({ day: -1 });

Creo que con esas adiciones, puedes hacer coincidir la velocidad de MySQL, incluso más rápido.

Además, mejor no lo use en tiempo real. Si sus datos no necesitan ser "minuciosamente" precisos, planifique una tarea de reducción de mapa de vez en cuando y utilice la colección de resultados.

Answer 4

extractos de la Guía Definitiva de MongoDB de O''Reilly:

El precio de usar MapReduce es la velocidad: el grupo no es particularmente rápido, pero MapReduce es más lento y no debe usarse en "tiempo real". Ejecuta MapReduce como un trabajo en segundo plano, crea una colección de resultados y luego puede consultar esa colección en tiempo real.

options for map/reduce:

"keeptemp" : boolean 
If the temporary result collection should be saved when the connection is closed. 

"output" : string 
Name for the output collection. Setting this option implies keeptemp : true.