Hadoop - MapReduce

MapReduce es un marco con la que podemos escribir aplicaciones para procesar grandes cantidades de datos, paralelamente, en grandes grupos de componentes de hardware de manera confiable.

MapReduce ¿Qué es?

MapReduce es una técnica de procesamiento y un programa modelo de computación distribuida basada en java. El algoritmo MapReduce contiene dos tareas importantes, a saber Mapa y reducir. Mapa toma un conjunto de datos y se convierte en otro conjunto de datos, en el que los elementos se dividen en tuplas (pares clave/valor). En segundo lugar, reducir tarea, que toma la salida de un mapa como entrada y combina los datos tuplas en un conjunto más pequeño de tuplas. Como la secuencia de MapReduce el nombre implica, la reducción se realiza siempre después de que el mapa.

La principal ventaja de MapReduce es que es fácil de escalar procesamiento de datos en múltiples nodos. En el modelo MapReduce, el procesamiento de datos primitivos son llamados mapas y reductores. Descomposición de una aplicación de procesamiento de datos en mapas y reductores a veces es no trivial. Pero, una vez que escribir una aplicación en el MapReduce forma, la escala de la aplicación que se ejecuta en cientos, miles o incluso decenas de miles de máquinas en un clúster es simplemente un cambio de configuración. Esta escalabilidad sencilla es lo que ha atraído a muchos programadores a usar el modelo MapReduce.

El algoritmo

• Paradigma MapReduce por lo general se basa en enviar el ordenador a donde residen los datos.

• MapReduce programa se ejecuta en tres etapas, a saber: mapa etapa, shuffle, y reducir.

  • Mapa etapa : El mapa o mapa de trabajo es la de procesar los datos de entrada. Por lo general, los datos de entrada se encuentra en la forma de archivo o directorio y se almacena en el sistema de archivos Hadoop (HDFS). El archivo de entrada se pasa a la función mapa línea por línea. El mapper procesa los datos y crea varios pequeños fragmentos de datos.

  • Reducir etapa: Esta etapa es la combinación de la reproducción aleatoria y la etapa Reducir. La pieza de trabajo es la de procesar los datos que llegan desde el mapa. Después de un proceso de elaboración, se genera un nuevo conjunto de la producción, que se almacena en el HDFS.

• Durante el trabajo, Hadoop MapReduce envía el mapa y reducir las tareas a los servidores correspondientes en el clúster.

• El marco de trabajo administra todos los detalles de los datos de tareas tales como la emisión, verificar la realización de las tareas, y copia de los datos en todo el clúster entre los nodos.

• La mayoría de los informáticos se lleva a cabo en los nodos de datos en los discos locales que reduce el tráfico de la red.

• Tras la finalización de las tareas asignadas, el grupo recopila y disminución de los datos que forman un resultado adecuado y lo envía de vuelta al servidor Hadoop.

Las entradas y salidas (Java)

El MapReduce marco funciona en <key, value> pares, es decir, el marco considera que la entrada en el trabajo como un conjunto de <key, value> pares y produce un conjunto de etiquetas <key, value> de la salida del trabajo, posiblemente de distintos tipos.

La clave y el valor de las clases deben ser serializados por el marco y por lo tanto, necesidad de aplicar la Escritura interfaz. Además, las clases clave tienen que aplicar la interfaz para facilitar Writable-Comparable ordenar por el marco. Tipos de entrada y salida de MapReduce trabajo: (Entrada)<k1, v1> -> map -> <k2, v2>-> reduce -> <k3, v3>(salida).

Terminología

• Carga útil - Aplicaciones aplicar el mapa y reduzca las funciones y forman el núcleo del trabajo.

• Mapa - Mapa mapas la entrada pares clave/valor a un conjunto de insumos par clave/valor.

• NamedNode - nodo que gestiona el Hadoop Distributed File System (HDFS).

• DataNode - Nodo en el que se presentan los datos antes de cualquier transformación.

• MasterNode JobTracker - Nodo donde se ejecuta y que acepta las peticiones de trabajo de los clientes.

• SlaveNode - Nodo en Mapa y Reducir programa se ejecuta.

• JobTracker - Los trabajos y las vías el asignar trabajos a Tarea tracker.

• Tarea Tracker - Realiza un seguimiento de la tarea y el estado de los informes de JobTracker.

• Trabajo: un programa es una ejecución de un mapa y reductor en un conjunto de datos.

• Tarea: La ejecución de un mapa o un reductor en un corte de datos.

• Tarea intento - un caso particular de un intento de ejecutar una tarea en un SlaveNode.

Escenario de ejemplo

A continuación se muestra los datos relativos al consumo eléctrico de una organización. Contiene el consumo eléctrico mensual y el promedio anual de diferentes años.

Si los datos anteriores no se da como entrada, tenemos que escribir aplicaciones para procesar y producir resultados tales como encontrar el año de uso máximo, año de uso mínimo, y así sucesivamente. Este es un paseo para los programadores con número finito de registros. Simplemente se escriben la lógica para producir los resultados requeridos, y pasar los datos a la aplicación escrita.

Sin embargo, creo que de los datos que representa el consumo eléctrico de todos los grandes sectores de un estado en particular, desde su formación.

Al escribir aplicaciones para procesar esos datos por lotes,

• Que tendrá un montón de tiempo para que se ejecute.
• Habrá una gran cantidad de tráfico en la red cuando nos movemos los datos desde el origen en el servidor de la red y así sucesivamente.

Para resolver estos problemas, tenemos el MapReduce framework.

Datos de entrada

Los datos anteriores se guardan como ejemplo.txtand dada como entrada. El archivo de entrada se ve como se muestra a continuación.

1979   23   23   2   43   24   25   26   26   26   26   25   26  25 
1980   26   27   28  28   28   30   31   31   31   30   30   30  29 
1981   31   32   32  32   33   34   35   36   36   34   34   34  34 
1984   39   38   39  39   39   41   42   43   40   39   38   38  40 
1985   38   39   39  39   39   41   41   41   00   40   39   39  45 

Programa de ejemplo

A continuación se muestra el programa de la muestra los datos usando MapReduce framework.

package hadoop; 

import java.util.*; 

import java.io.IOException; 
import java.io.IOException; 

import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapred.*; 
import org.apache.hadoop.util.*; 

public class ProcessUnits 
{ 
   //Mapper class 
   public static class E_EMapper extends MapReduceBase implements 
   Mapper<LongWritable ,/*Input key Type */ 
   Text,                /*Input value Type*/ 
   Text,                /*Output key Type*/ 
   IntWritable>        /*Output value Type*/ 
   { 
      
      //Map function 
      public void map(LongWritable key, Text value, 
      OutputCollector<Text, IntWritable> output,   
      Reporter reporter) throws IOException 
      { 
         String line = value.toString(); 
         String lasttoken = null; 
         StringTokenizer s = new StringTokenizer(line,"\t"); 
         String year = s.nextToken(); 
         
         while(s.hasMoreTokens())
            {
               lasttoken=s.nextToken();
            } 
            
         int avgprice = Integer.parseInt(lasttoken); 
         output.collect(new Text(year), new IntWritable(avgprice)); 
      } 
   } 
   
   
   //Reducer class 
   public static class E_EReduce extends MapReduceBase implements 
   Reducer< Text, IntWritable, Text, IntWritable > 
   {  
   
      //Reduce function 
      public void reduce( Text key, Iterator <IntWritable> values, 
         OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException 
         { 
            int maxavg=30; 
            int val=Integer.MIN_VALUE; 
            
            while (values.hasNext()) 
            { 
               if((val=values.next().get())>maxavg) 
               { 
                  output.collect(key, new IntWritable(val)); 
               } 
            } 
 
         } 
   }  
   
   
   //Main function 
   public static void main(String args[])throws Exception 
   { 
      JobConf conf = new JobConf(Eleunits.class); 
      
      conf.setJobName("max_eletricityunits"); 
      conf.setOutputKeyClass(Text.class);
      conf.setOutputValueClass(IntWritable.class); 
      conf.setMapperClass(E_EMapper.class); 
      conf.setCombinerClass(E_EReduce.class); 
      conf.setReducerClass(E_EReduce.class); 
      conf.setInputFormat(TextInputFormat.class); 
      conf.setOutputFormat(TextOutputFormat.class); 
      
      FileInputFormat.setInputPaths(conf, new Path(args[0])); 
      FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
      
      JobClient.runJob(conf); 
   } 
} 

Guardar el programa comoProcessUnits.java.La compilación y la ejecución del programa se explica a continuación.

Compilación y Ejecución del Programa Unidades de proceso

Supongamos que estamos en el directorio principal de un usuario Hadoop (e.g. /home/hadoop).

Siga los pasos que se indican a continuación para compilar y ejecutar el programa anterior.

Paso 1

El comando siguiente es crear un directorio para almacenar el compilado las clases de java.

$ mkdir units 

Paso 2

DescargarHadoop-core-1.2.1.jar,que se utiliza para compilar y ejecutar el programa MapReduce. Visita el siguiente enlace http://mvnrepository.com/artifact/org.apache.hadoop/hadoop-core/1.2.1 para descargar el jar. Supongamos que el archivo .exe descargado es/home/hadoop/.

Paso 3

Los siguientes comandos se utilizan para la elaboración de lasProcessUnits.javaprograma java y crear una jarra para el programa.

$ javac -classpath hadoop-core-1.2.1.jar -d units ProcessUnits.java 
$ jar -cvf units.jar -C units/ . 

Paso 4

El siguiente comando se utiliza para crear una entrada de directorio HDFS.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir 

Paso 5

El siguiente comando se utiliza para copiar el archivo de entrada llamadosample.txtel directorio de entrada de HDFS.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/sample.txt input_dir 

Paso 6

El siguiente comando se utiliza para verificar los archivos en el directorio de entrada.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/ 

Paso 7

El siguiente comando se utiliza para ejecutar la aplicación Eleunit_max teniendo en los archivos de entrada en el directorio de entrada.

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir 

Espere un momento hasta que el archivo es ejecutado. Después de la ejecución, como se muestra a continuación, la salida contendrá el número de divisiones, el número de tareas mapa, el número de pieza tareas, etc.

INFO mapreduce.Job: Job job_1414748220717_0002 
completed successfully 
14/10/31 06:02:52 
INFO mapreduce.Job: Counters: 49 
File System Counters 
 
FILE: Number of bytes read=61 
FILE: Number of bytes written=279400 
FILE: Number of read operations=0 
FILE: Number of large read operations=0   
FILE: Number of write operations=0 
HDFS: Number of bytes read=546 
HDFS: Number of bytes written=40 
HDFS: Number of read operations=9 
HDFS: Number of large read operations=0 
HDFS: Number of write operations=2 Job Counters 


   Launched map tasks=2  
   Launched reduce tasks=1 
   Data-local map tasks=2  
   Total time spent by all maps in occupied slots (ms)=146137 
   Total time spent by all reduces in occupied slots (ms)=441   
   Total time spent by all map tasks (ms)=14613 
   Total time spent by all reduce tasks (ms)=44120 
   Total vcore-seconds taken by all map tasks=146137 
   
   Total vcore-seconds taken by all reduce tasks=44120 
   Total megabyte-seconds taken by all map tasks=149644288 
   Total megabyte-seconds taken by all reduce tasks=45178880 
   
Map-Reduce Framework 
 
Map input records=5  
   Map output records=5   
   Map output bytes=45  
   Map output materialized bytes=67  
   Input split bytes=208 
   Combine input records=5  
   Combine output records=5 
   Reduce input groups=5  
   Reduce shuffle bytes=6  
   Reduce input records=5  
   Reduce output records=5  
   Spilled Records=10  
   Shuffled Maps =2  
   Failed Shuffles=0  
   Merged Map outputs=2  
   GC time elapsed (ms)=948  
   CPU time spent (ms)=5160  
   Physical memory (bytes) snapshot=47749120  
   Virtual memory (bytes) snapshot=2899349504  
   Total committed heap usage (bytes)=277684224
     
File Output Format Counters 
 
   Bytes Written=40 

Paso 8

El siguiente comando se utiliza para verificar los archivos resultantes de la carpeta de salida.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/ 

Paso 9

El siguiente comando se utiliza para ver el resultado en la Part-00000 archivo. Este archivo se genera por HDFS.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000 

A continuación se muestra la salida generada por el MapReduce programa.

1981    34 
1984    40 
1985    45 

Paso 10

El siguiente comando se utiliza para copiar la carpeta de salida de los HDFS en el sistema de archivos local para analizar.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000/bin/hadoop dfs get output_dir /home/hadoop 

Comandos importantes

Hadoop Todos los comandos son invocados por el$HADOOP_HOME/bin/hadoop. Ejecuta el script Hadoop sin argumentos imprime la descripción de todos los comandos.

Usage : hadoop [--config confdir] COMANDO

La siguiente tabla detalla las opciones disponibles y su descripción.

Cómo interactuar con los trabajos MapReduce

Uso: hadoop job [GENERIC_OPTIONS]

Las siguientes son las opciones genéricas disponibles en Hadoop trabajo.

Para ver el estado de trabajo

$ $HADOOP_HOME/bin/hadoop job -status <JOB-ID> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -status job_201310191043_0004 

Para ver la historia de salida del trabajo-dir

$ $HADOOP_HOME/bin/hadoop job -history <DIR-NAME> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -history /user/expert/output 

Para matar el trabajo

$ $HADOOP_HOME/bin/hadoop job -kill <JOB-ID> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -kill job_201310191043_0004