MapReduce
| ||||
MapReduce, es un modelo, framework de programación utilizado para dar soporte a la computación paralela sobre grandes colecciones de datos en grupos de computadoras y al commodity computing. Su nombre está inspirado en los nombres de dos importantes métodos, macros o funciones en programación funcional: Map y Reduce.
MapReduce ha sido adoptado mundialmente, ya que existe una implementación Open Source denominada Hadoop. Su desarrollo fue liderado inicialmente por Yahoo y actualmente lo realiza el proyecto Apache. Se han escrito implementaciones de bibliotecas de MapReduce en diversos lenguaje de programación como C++, Java y Python.
MapReduce consiste en un solo JobTracker maestro y un TaskTracker esclavo por nodo de clúster. El maestro es responsable de programar las tareas de los componentes de los trabajos en los esclavos, supervisarlos y volver a ejecutar las tareas fallidas. Los esclavos ejecutan las tareas según las instrucciones del maestro.
Sumario
Historia
Las primeras implementaciones de Google necesitaban realizar operaciones de multiplicación de grandes matrices para calcular el PageRank, o lo que es lo mismo el ranking de páginas en una búsqueda. De esta forma se hizo popular MapReduce como un método de cálculo de álgebra lineal. Por tratar grandes colecciones de datos, llevó a crear algoritmos y frameworks capaces de poder procesar terabytes de información. Una de las primeras aplicaciones capaces de programar MapReduce fue implementado inicialmente en Hadoop, diseñado inicialmente por Doug Cutting, que lo nombró así por su elefante de juguete. Fue desarrollado originalmente para apoyar la distribución del proyecto de motor de búsqueda Nutch.
Características
Apache MapReduce es un poderoso marco para procesar grandes conjuntos distribuidos de datos estructurados o no estructurados en un clúster Hadoop. Entre sus características están:
- Su capacidad para realizar el procesamiento a través de un clúster entero de nodos, con cada nodo procesando sus datos locales.
MapReduce abstrae la complejidad de la programación distribuida, permitiendo a los programadores describir el procesamiento que les gustaría realizar en términos de una función de mapa y una función de reducción. En el momento de la ejecución, durante la fase de mapa, varios nodos en el clúster, llamados mapeadores, leen en datos crudos locales en pares clave-valor. A esto le sigue una fase de ordenación y aleatorización, en la que cada mapeador ordena sus resultados mediante teclas y remite los rangos de teclas a otros nodos del grupo, llamados reductores. Finalmente, en la fase de reducción, los reductores analizan los datos de las claves que pasaron de los mapeadores. MapReduce v1, incluido en todas las versiones de MapR Distribution, tiene dos propósitos en el clúster de Hadoop. En primer lugar, MapReduce actúa como gestor de recursos para los nodos del clúster Hadoop. Emplea un JobTracker para dividir un trabajo en múltiples tareas, distribuir y supervisar su progreso a uno o más TaskTrackers, que realizan el trabajo en paralelo. Como gestor de recursos, es un componente clave del clúster, que sirve como plataforma para muchas aplicaciones Hadoop de nivel superior, incluyendo Pig (link) y Hive (enlace).
- MapReduce, sirve como un motor de procesamiento de datos, la ejecución de puestos de trabajo que se expresan con el mapa y reducir la semántica.
Comenzando con la versión MapR 4.0, MapR incluye MapReduce v2 además de la v1. MapReduce v2 fue rediseñado para funcionar sólo como un motor de procesamiento de datos, haciendo girar la funcionalidad del gestor de recursos en un nuevo componente llamado YARN (Yet Another Resource Negotiator) ( link ). Antes de esta división, las aplicaciones de nivel superior que requerían acceso a los recursos de Hadoop tenían que expresar sus trabajos usando mapa y reducir la semántica, con cada trabajo pasando por el mapa, ordenar, mezclar, reducir procesos. Esto no era adecuado para algunos tipos de trabajos que no encajaban bien en el paradigma de MapReduce, ya sea porque requerían tiempos de respuesta más rápidos de lo que permitiría un ciclo MapReduce completo, o porque requerían un procesamiento más complejo que el que no podía expresarse en MapReduce único Trabajos, como el procesamiento de gráficos. Con YARN, los clústeres Hadoop se vuelven mucho más versátiles, permitiendo que el mismo clúster se utilice tanto para procesamiento clásico de MapReduce por lotes como para trabajos interactivos como SQL.
Entradas y salidas
El framework MapReduce opera exclusivamente con pares <key, value> , es decir, el framework visualiza la entrada al trabajo como un conjunto de pares <key, value> y produce un conjunto de pares <key, value> como salida del Trabajo, concebible de diversos tipos.
La clave y las clases de valor tienen que ser serializable por el marco y por lo tanto, la necesidad de implementar la interfaz de escritura. Además, las clases clave tienen que implementar la interfaz WritableComparable para facilitar la clasificación por el marco.
Tipos de entrada y salida de un trabajo MapReduce:
(Entrada) <k1, v1> -> mapa -> <k2, v2> -> combinar -> <k2, v2> -> reducir -> <k3, v3>.
Funciones
- Función Map()
Map toma uno de estos pares de datos con un tipo en un dominio de datos, y devuelve una lista de pares en un dominio diferente:
Map(k1,v1) -> list(k2,v2).
La función map(): se encarga del mapeo y es aplicada en paralelo para cada ítem en la entrada de datos. Esto produce una lista de pares (k2,v2) por cada llamada. Después de eso, el framework de MapReduce junta todos los pares con la misma clave de todas las listas y los agrupa, creando un grupo por cada una de las diferentes claves generadas. Desde el punto de vista arquitectural el nodo master toma el input, lo divide en pequeñas piezas o problemas de menor identidad, y los distribuye a los denominados worker nodes. Un worker node puede volver a sub-dividir, dando lugar a una estructura arbórea. El worker node procesa el problema y pasa la respuesta al nodo maestro.
- Función Reduce()
La función reduce es aplicada en paralelo para cada grupo, produciendo una colección de valores para cada dominio:
Reduce(k2, list (v2)) -> list(v3).
La función reduce(): cada llamada a Reduce típicamente produce un valor v3 o una llamada vacía, aunque una llamada puede retornar más de un valor. El retorno de todas esas llamadas se recoge como la lista de resultado deseado.
Por lo tanto, el framework MapReduce transforma una lista de pares (clave, valor) en una lista de valores. Este comportamiento es diferente de la combinación "map and reduce" de programación funcional, que acepta una lista arbitraria de valores y devuelve un valor único que combina todos los valores devueltos por mapa.
Arquitectura del MapReduce
La función map() se ejecuta de forma distribuida a lo largo de varias máquinas. Los datos de entrada, procedentes por regla general de un gran archivo (fichero), se dividen en un conjunto de M particiones de entrada de generalmente 16 megabytes. Estas particiones pueden ser procesadas en diversas máquinas. En una invocación de MapReduce suelen ocurrir varias operaciones:
- Se procede a dividir las entradas en M particiones de tamaño aproximado de 64 megabytes. El programa MapReduce se comienza a instanciar en las diversas máquinas del cluster. Por regla general, el número de instancias se configura en las aplicaciones.
- Una de las copias del programa es especial y toma el papel de "maestro". El resto de copias se denominan como "workers" y reciben la asignación de sus tareas desde el master. Se considera que existen una cantidad de M map() tareas y de R reduce(). El "maestro" se encarga de recopilar "workers" en reposo (es decir sin tarea asignada) y le asignará una tarea específica de map() o de reduce(). Un worker sólo puede tener tres estados: reposo, trabajando, completo.
- Un worker que tenga asignada una tarea específica de map() tomará como entrada la partición que le corresponda. Se dedicará a parsear los pares (clave, valor) para crear una nueva pareja de salida, tal y como se especifica en su programación. Los pares clave y valor producidos por la función map() se almacenan como buffer en la memoria.
- Periódicamente, los pares clave-valor almacenados en el buffer se escriben en el disco local, repartidos en R regiones. Las regiones de estos pares clave-valor son pasados al master, que es responsable de redirigir a los "workers" que tienen tareas de reduce().
- Cuando un worker de tipo reduce es notificado por el "maestro" con la localización de una partición, éste emplea llamadas remotas para hacer lecturas de la información almacenada en los discos duros de los diversos workers de tipo map(). cuando un worker de tipo reduce() lee todos los datos intermedios, ordena las claves de tal modo que a se agrupan los datos encontrados que poseen la misma clave. El ordenamiento es necesario debido a que, por regla general, muchas claves de funciones map() diversas pueden ir a una misma función reduce(). En aquellos casos en los que la cantidad de datos intermedios sean muy grandes, se suele emplear un ordenamiento externo.
- El worker de tipo reduce() itera sobre el conjunto de valores ordenados intermedios, y lo hace por cada una de las claves únicas encontradas. Toma la clave y el conjunto de valores asociados a ella y se los pasa a la función reduce(). La salida de reduce() se añade al archivo (fichero) de salida de MapReduce.
- Cuando todas las tareas map() y reduce() se han completado, el "maestro" levanta al programa del usuario. Llegados a este punto la llamada MapReduce retorna el control al código de un usuario.
Se considera que ha habido un final de las tareas cuando este control se ha devuelto al usuario. Las salidas se ditribuyen en un fichero completo, o en su defecto se reparten en R ficheros. Estos R ficheros pueden ser la entrada de otro MapReduce o puede ser procesado por cualquier otro programa que necesite estos datos.
Combinador (Agregadores locales)
En un entorno de clusterización, uno de las límitaciones se encuentra en el transporte de grandes ficheros entre ordenadores que debido a lo limitado de su ancho de banda. En el framework MapReduce la función map() escribe en una memoria intermedia de caracter local, como puede ser un disco duro. La información que se escribe en local es agregada y ordenada por una función agregadora encargada de realizar esta operación. Los valores ordenados son de la forma [k, [v1, v2, v3,] ..., vn]]. De esta forma la función reduce() recibe una lista de valores asociados a una única clave procedente del combinador. Debido a que la latencia de red de ordenadores, y de sus discos suele ser mayor que cualquier otra de las operaciones, cualquier reducción en la cantidad de datos intermedios incrementará la eficiencia de los algoritmos. En MapReduce, cualquier agregación local de los resultados intermedios causa una mejora real de la eficiencia global. Es por esta razón por la que muchas ditribuciones oficiales de MapReduce suelen incluir operaciones de agregación en local, mediante el uso de funciones capaces de agregar datos localmente. Evitando, o reduciendo en la medida de lo posible el movimiento de grandes ficheros. Bien sea añadidas a las funciones map(), o a los agregadores locales.
Tolerancia a Fallos
El mecanismo de MapReduce es tolerante a fallos cuando uno de los workers se ve sometido a un fallo. Como MapReduce se ha diseñado para procesos en los que se encuentran involucrados grandes tamaños de datos mediante el empleo de cientos o miles de ordenadores. Aún siendo la probabilidad de fallo baja, es muy posible que uno (o varios) de los workers quede desactivo precisamente por fallo de la máquina que le daba soporte. El "master" periódicamente hace ping a cada worker para comprobar su estatus. Si no existe respuesta tras un cierto instante de espera, el master interpreta que el worker está desactivado. Cualquier tarea map() que ha sido completa por el worker regresa de inmediato a su estado de espera, y por lo tanto puede resultar elegible para su asignación en otros workers. De forma similar, cualquier función map() (o reduce) que se encuentre en progreso durante el fallo, se resetea a estado de reposo pudiendo ser elegida para su nueva re-asignación. Las tareas de map() completados se vuelven a re-ejecutar ante un fallo debido en parte a que su salida se almacena en los discos locales de la máquina que falló, y por lo tanto se consideran inaccesibles. Las tareas reduce() completas no son necesarias volver a ser re-ejecutadas debido a que su salida se ha almacenado en el sistema global. cuando la tarea de map() se ejecuta por un worker A y luego por un worker B (debido principalmente a un fallo), en este caso todas las tareas reduce() son notificadas para que eliminen datos procedentes del worker A y acepten las del worker B. De esta forma la ejecucción de MapReduce es resiliente.
Ejemplos
En la descripción de los ejemplos de uso de MAPREDUCE sólo es necesario describir en detalle como se implementan las operaciones de map() y de reduce() en cada caso. La literatura muestra ejemplos reiterados de conteo de palabras en un documento, de operaciones matriciales y de operaciones de consulta a bases de datos relacionales.
Conteo de palabras
Este ejemplo de MAPREDUCE es un proceso para contar las apariciones de cada palabra en un conjunto de documentos:
map(String name, String document): // clave: nombre del documento // valor: contenido del documento for each word w in document: EmitIntermediate(w, 1);
La función map() en este caso divide un documento en palabras (es decir lo tokeniza) mediante el empleo de un simple analizador léxico, y emite una serie de tuplas de la foma (clave, valor) donde la clave es la palabra y el valor es "1". Es decir, por ejemplo, del documento "La casa de la pradera" la función map retornaría: ("la", "1"), ("casa", "1"), ("de", "1"), ("la", "1"), ("pradera", "1").
reduce(String word, Iterator partialCounts): // word: una palabra // partialCounts: una lista parcial para realizar cuentas agregadas int result = 0; for each v in partialCounts: result += ParseInt(v); Emit(result);
Aquí, cada documento es dividido en palabras, y cada palabra se cuenta con valor inicial "1" por la función Map, utilizando la palabra como el resultado clave. El framework reúne todos los pares con la misma clave y se alimenta a la misma llamada Reduce, por lo tanto, esta función sólo necesita la suma de todos los valores de su entrada para encontrar el total de las apariciones de esa palabra. En el ejemplo anterior ("la", "1") aparece dos veces debido a que la clave "la" tiene dos ocurrencias, el resto de claves sólo aparece una vez.
Multiplicación de una matriz por un vector
Los ejemplos de algebra lineal para operaciones de matrices son los más adecuados por la idonidad del framework en estos casos. Supongamos que tenemos una matriz cuadrada M de tamaño nxn. Al elemento ubicado en la fila i y columna j le denominamos mij. Supongamos que tenemos un vector v de tal forma que en la posición j se tiene el elemento vj. De esta forma la resultante de la multiplicación entre la matriz M y el vector v será un vector x de longitud n, de tal forma que el elemento xi es tal que:
Esta operación se realiza sin problema alguno para matrices de varios miles de elementos, siendo costoso para varios millones. El problema de su computación proviene cuando se pretende realizar con centenares de billones. Es por esta razón por la que se asume en la aplicación de MAPREDUCE que n es del orden de 1012. La función map () en este caso toma una fila i de la matriz y completo el vector v para formar pares: (i, mijvj). Es decir de la forma (1, m11v1), (1, m12v2), (1, mi3v3) ... (1, mijvj).
map(Vector rowMatrix, Vector vector): // clave: i -> índice del vector // valor: producto de mij por vj. for each position i in vector: EmitIntermediate(i, value);
La función reduce() en este caso sólo tiene que colectar los pares que poseen la misma clave i y sumarlos.
reduce(String word, Iterator partialCounts): // word: una palabra // partialCounts: una lista parcial para realizar cuentas agregadas int result = 0; for each v in partialCounts: result += ParseInt(v); Emit(result);
Contribuyente (s): Stephen J. Bigelow y Mark C. Chu-Carroll MapReduce es un componente básico del framework de software Apache Hadoop . Hadoop permite el procesamiento resiliente y distribuido de conjuntos masivos de datos no estructurados a través de clusters de ordenadores de productos básicos, en los que cada nodo del clúster incluye su propio almacenamiento. MapReduce sirve dos funciones esenciales: Distribuye el trabajo a varios nodos dentro del cluster o mapa y organiza y reduce los resultados de cada nodo en una respuesta cohesiva a una consulta.
Componentes
MapReduce se compone de varios componentes, incluyendo:
- JobTracker: el nodo maestro que gestiona todos los trabajos y recursos de un clúster.
- TaskTrackers: agentes desplegados en cada máquina del clúster para ejecutar el mapa y reducir las tareas.
- JobHistoryServer: componente que rastrea los trabajos terminados, y se despliega típicamente como una función separada o con JobTracker.
Localidad
MapReduce se construye en la parte superior de GFS, el sistema de archivos de Google. Los archivos de entrada y salida se almacenan en GFS. Los trabajadores de MapReduce se ejecutan en servidores chunkservers de GFS. El maestro MapReduce intenta programar un trabajador de mapa en una de las máquinas que contiene una copia del fragmento de entrada que necesita para procesar. Alternativamente, MapReduce puede leer o escribir en BigTable.
