Primeros pasos de MapReduce con Hadoop

3
29982

Primeros pasos de MapReduce con Hadoop

0. Índice de contenidos.


1. Introducción.

Una vez visto el tutorial de primeros pasos con Hadoop, instalación y configuración, muchos se dirán: «pues muy bonito, ¿y esto para qué vale?». Como dijimos en el anterior tutorial, principalmente para realizar tareas distribuidas sobre grandes ficheros de datos utilizando el paradigma MapReduce. En este tutorial vamos a ponernos manos a la obra realizando un ejemplo simple de un algoritmo MapReduce con el API de Hadoop.

Como aquí lo importante son los datos y dado que el clásico wordcount (el holamundo de MapReduce) está muy visto, he buscado un dataset que pueda darme un poco más de miga para sacar un poco de jugo a los datos.

Por suerte, aunque queda mucho camino por recorrer, nuestras administraciones van publicando lo que se conoce como open data o datos abiertos poniendo a disposición del que quiera y de forma libre datos relativos a censos, información económica, laboral, sectorial, mediciones medioambientales, datos estadísticos de consumo, transportes, etc. A mí me ha parecido interesante un dataset sobre la medición de la calidad del aire en Castilla y León desde 1997 que he sacado de la página datosabiertos.jcyl.es . Vamos a ver con Hadoop cómo extraer información relevante de un fichero que contiene más de 150.000 líneas.

Puedes descargarte el código del tutorial desde mi repositorio de github pinchando aquí.

2. Entorno.

El tutorial se ha realizado con el siguiente entorno:

  • Ubuntu 12.04 64 bits
  • Oracle Java SDK 1.6.0_27
  • Apache Hadoop 2.2.0
  • Apache Maven 3.1.1

3. Configuración del proyecto

Lo primero que haremos será crear un proyecto Maven y añadir al pom.xml las dependencias necesarias para trabajar.

4. Análisis del DataSet

Antes de implementar ni una línea de código debemos pararnos a analizar lo que pretendemos extraer del fichero de datos. Si abrimos el fichero CSV que nos hemos descargado esto es lo que nos aparece.

Como se puede ver, el fichero tiene una línea por cada día de medición de diferentes medidas relativas a diferentes contaminantes como son el monóxido de carbono, monóxido de nitrógeno, dióxido de nitrógeno, ozono, etc… por provincia y población donde se encuentra la estación donde se ha tomado la medida.

Para hacer una primera prueba sencilla con el API de Hadoop vamos a calcular la media de las muestras tomadas desde 1997 de Monóxido de Carbono (CO) agrupados por provincia. Este sería un cálculo que aporta valor, podemos saber qué provincia de Castilla y León tiene un menor nivel de presencia de esta sustancia. Sin ser ningún entendido en química la he escogido por ser una sustancia que se desprende al quemar gasolina, petróleo, carbón, etc. y que representan un problema por estar presentes en el aire que respiramos.

El ranking final de las provincias menos contaminadas que resulte del cálculo que vamos a realizar, no se debe tomar como un estudio serio para determinar la calidad del aire presente en ellas ya que se ha decidido tomar este indicador como podría haber sido cualquier otro. Imagino que el cálculo para medir en conjunto la calidad del aire de una provincia debe ser mucho más complejo pero queda fuera del ámbito de este tutorial.

5. Mapper

Dentro del paradigma MapReduce, en la tarea mapper se reciben los bloques de datos que contienen la información. Es aquí donde debemos extraer la información que nos irá llegando línea a línea. Nuestra clase AirQualityMapper extenderá de la clase Mapper definiendo los formatos de entrada de la clave y valor y los tipos de salida que devolverá la función. La salida del mapper será la entrada del reducer. En nuestro caso el valor de la clave contendrá el offset del fichero es decir el puntero que va recorriendo el fichero entero y extrae cada una de las líneas del mismo. La línea nos la pasará en el campo value. Al heredar de Mapper debemos implementar el método map cuyos parámetros de entrada corresponderán con los tipos genéricos la que definimos anteriormente.

El método map es muy sencillo, considerando que en el campo value tenemos una línea del fichero de entrada, la troceamos separando por el token «;». Nos quedaremos con los valores que nos interesan, para nuestro análisis nos quedamos con el valor correspondiente al Monóxido de Carbono (CO) y con la provincia.

Aparte de la tarea de extracción del dato, el mapper también realiza la tarea de filtrado, en nuestro caso no emitimos ningún valor que no sea numérico. Hacemos esto porque hay líneas que no tienen todas las medidas bien informadas supongo que porque ese día la muestra tomada no es fiable de acuerdo a las directiva que regula la medición.

Una parte muy importante en la fase de map es la información que devuelve el propio método, como es lógico. Esto se realiza escribiendo en el objeto context siempre una tupla compuesta por una clave y un valor. La clave será la provincia y como valor el dato relativo a la medición de Monóxido de Carbono (CO). Los tipos que se escriben en el context son del tipo Writable. Es un tipo de datos específico en Hadoop ya que por temas de rendimiento redefine los tipos de Java, por ejemplo si la clave es de tipo string tenemos que crear un tipo Text, si el valor es de tipo Double, debemos crear un DoubleWritable. También podremos crear nuestros propios Writables para trabajar con objetos compuestos. Esto lo veremos en próximos tutoriales.

El código del mapper quedaría así:

6. Reducer

El reducer es la parte del algoritmo que recoge agrupadas por clave todos los valores emitidos en la fase map. En nuestro caso a la tarea reduce le llegará algo como (ÁVILA, [1.2, 1.3, 1.5, 1.6, 1.4…]). En esta fase debemos realizar la operación que calcule la media de los valores recibidos para la provincia.

De forma muy parecida al map, en este caso extendemos de Reducer y definimos los valores de entrada y de salida del método reduce que debemos implementar obligatoriamente. Los valores de entrada deben coincidir con los valores de salida de la función map. La salida del reduce será de tipo texto resultando la tupla que contendrá la provincia y el valor medio de las muestras tomadas de Monóxido de Carbono (CO) desde 1997.

Recorremos con un bucle for el iterable que Hadoop nos pasa con los valores de las muestras tomadas. En el campo key vendrá la provincia. Si existen medidas para la provincia se emitirá el resultado escribiendo en el objeto context. De forma similar al map, se escribe siempre una tupla key/value. En nuestro caso la provincia y la media obtenida redondeando en 2 decimales.

7. Driver

Pues ahora que ya tenemos todo, sólo nos faltaría ejecutarlo. Para eso nos creamos una función Driver que no es más que un método main que creará un Job configurando todo lo necesario para ejecutar la tarea: función map, función reduce, tipos de datos que manejan, entrada y salida de datos, etc. Utilizaremos también la clase ToolRunner de Hadoop, una utilidad que se encarga de pasar en la invocación por línea de comandos de nuestra clase Driver una serie de parámetros de configuración que le pasemos.

El resultado final (eliminando código repetido) quedaría así:

Creamos la clase AirQualityManager que para poder invocarse a través de ToolRunner debe extender de Configured e implementar el interfaz Tool. Define el método main que servirá de punto de entrada a nuestra función. En el main invocamos al ToolRunner creando una instancia de nuestra clase manager pasándole los argumentos que recogemos por línea de comandos.

Esto llamará a nuestro método run que valida los datos recibidos como entrada ya que es necesario introducir el fichero que contiene los datos y también el directorio de salida donde se escribirá el resultado. A continuación borra el directorio si ya existe (Hadoop da un error si ya está creado) y crea un nuevo Job pasándole la configuración recogida por línea de comandos si la hubiera. Al job hay que configurarle una serie de parámetros como son el formato de entrada y salida, en nuestro caso una entrada de texto línea a línea, el mapper y el reducer, los tipos de datos que se manejan, los ficheros de entrada y el directorio de salida. Invocamos y esperamos mediante la llamada al método waitForCompletion. El valor true indica el modo verbose.

Ya sólo queda ejecutar y esperar el resultado. Para ello antes debemos crear el directorio y subir al HDFS el fichero que contiene los datos mediante el comando:

Podemos comprobar si se ha subido bien mediante el comando:

O también desde la url http://localhost:50075 donde se ve la información del NameNode y del filesystem de HDFS.

El siguiente paso será generar un jar que será la forma de pasarle a Hadoop el código que tiene que ejecutar. Para ello usamos maven haciendo un mvn package. Una vez hecho esto ejecutamos el siguiente comando para lanzar el Job.

Después de unos pocos segundos nos dejará el resultado en el directorio output. Lo abrimos y tendremos el resultado del cálculo de la media de Monóxido de Carbono (CO) por provincia.

Enhorabuena a los sorianos por tener un aire tan limpio 🙂

8. Conclusiones.

En este tutorial hemos visto lo sencillo que resulta crear un Job en Hadoop y extraer información de un fichero que a simple vista no nos dice gran cosa a menos que se procese. El fichero con el que hemos trabajado era pequeño, unas 150.000 líneas (9 MB). Imaginad ficheros de cientos de millones de líneas, quizá la tarea no tardaría unos segundos en terminar sino horas o días por lo que ya tendríamos todo montado para distribuir nuestra tarea en un cluster que redujera los tiempos de proceso.

En próximos tutoriales complicaremos un poco más el problema y explotaremos un poquito más la información de la que disponemos.

Puedes descargarte el código del tutorial desde mi repositorio de github pinchando aquí.

Espero que te haya sido de ayuda.

Un saludo.

Juan

3 Comentarios

  1. Excelente tutorial, para que lo entienda un lerder como yo es significa que esta muy bien. Como pegas, deberia de venir la fecha del tutorial arriba porque ya sabemos como evoluciona estas tecnologias y en 2 años ya el tema queda desfasado por lo que deberiamos saberlos.
    En el tutorial viene que en proximos tutoriales nos contaran más pero no hay links a los siguientes tutoriales. ¿como lo busco? deberian de poner links de tutoriales relacionados.

    Gracias por su aportacion a la comunidad. son excelentes

  2. Hola amigo, soy muy nuevo en esto. Me pdrias hacer el favor de explicar los parametros que estan en Mapper
    en esta linea:
    public static class AirQualityMapper extends Mapper
    y tambien en esta:
    public static class AirQualityReducer extends Reducer
    GRACIAS.

Dejar respuesta

Please enter your comment!
Please enter your name here