Speculative Execution

La meta del modelo MapReduce es dividir trabajos en tareas más pequeñas y ejecutarlas en paralelo para que el tiempo de ejecución total del trabajo sea menor que si se ejecutaran de forma secuencial.

Muchas veces, para grandes cantidades de datos, el número de divisiones es mayor que el número de nodos en el cluster, así que aunque las tareas se van ejecutan de forma paralela, otras tareas tienen que esperar a que las primeras finalicen para poder continuar.

Si por alguna causa una de las tareas se está ejecutando más lentamente de lo normal se podrían producir cuellos de botella y el tiempo dedicado para la ejecución total podría aumentar de forma considerable con respecto a si todo el clúster estuviera funcionando de manera correcta.

Hadoop no se va a encargar de buscar la causa de qué es lo que está retardando una tarea, sino que va a buscar soluciones para mitigar esto, y la Speculative Execution es la técnica que utiliza.

En el momento que JobTracker detecta que una tarea es más lenta de lo normal, va a lanzar una tarea especulativa (speculative task), pero sólo en el momento en el que el resto de las tareas del trabajo han finalizado, que la tarea haya estado funcionando al menos un minuto o que el tiempo medio de ejecución de esa tarea sea más alto que el del resto de tareas. Esto es, va a lanzar un duplicado de la tarea que está funcionando lentamente y las tareas se van a estar ejecutando al mismo tiempo.

Cuando una de las tareas finaliza su ejecución, el JobTracker se encargará de desechar el otro proceso (hace un kill de la tarea).

Si el defecto en la tarea se trata de un bug en el programa, éste va a suceder también en la tarea especulativa, así que tendría que ser el desarrollador quien corrigiera ese bug.

La Speculative Execution está activada por defecto, pero está permitido desactivarla en los ficheros de configuración.
En general es conveniente desactivar la speculative execution para las tareas Reduce porque, por un lado, siempre que se duplica una tarea Reduce ésta tiene que recuperar los datos intermedios de la red, lo que incrementaría el tráfico en la red.
Y por otro lado porque si la distribución de las key que reciben los Reducer no es equilibrada, es normal que una tarea Reduce tarde más que otra, por ejemplo, si pensamos en el WordCount, en inglés, la palabra "the" va a tener un array de valores mucho mayor que otras key, por lo que será normal que esta tarea Reduce tarde más.

Por el contrario, como regla general, se recomienda activar la Speculative Execution para las tareas Map. Una excepción podría ser para tareas que no son idempotentes

Paso de Parámetros en Hadoop

Algo tan simple como un paso de parámetros entre el Driver y el Mapper o Reducer puede llevarnos a un error ya que como desarrolladores estamos acostumbrados a que pueda haber una cierta "comunicación" entre las clases.

En Hadoop esto no es correcto, y a estas alturas deberíamos tener claro que este tipo de aplicaciones se ejecuta de forma distribuída, que cada tarea se ejecuta en una JVM diferente e incluso en distintos nodos del clúster.

Voy a dejaros dos ejemplos, uno con la forma correcta y otro con la forma incorrecta por si queréis ver qué pasa, ya que forma la incorrecta no es que dé un error, el Job va a ejecutarse sin problemas, pero el parámetro no se pasará.

También hay que tener cuidado, ya que si se ejecuta en modo local, el paso de parámetros SÍ que funcionará. Sin embargo, no funcionará en el momento que llevemos nuestro programa al modo distribuído o pseudo-distribuído.

El ejemplo es muy simple y ni si quiera vamos a usar la función Reducer. A partir del fichero score.txt que ya he usado para otros ejemplos de la forma:

01-11-2012 Pepe Perez Gonzalez 21
01-11-2012 Ana Lopez Fernandez 14
15-02-2013 Angel Martin Hernandez 3
01-11-2012 Maria Garcia Martinez 11
01-11-2012 Pablo Sanchez Rodriguez 9
01-11-2012 Angel Martin Hernandez 3
15-01-2013 Pepe Perez Gonzalez 17
15-01-2013 Maria Garcia Martinez 3
...

Sólo queremos como salida la lista de personas con una puntuación mayor de 25, y ese valor lo queremos pasar desde el Driver.

Forma Incorrecta

 
public class PersonaScoreDriver {
 
 private static int score;
 
 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  //Indicamos el parámetro a pasar
  
  score = 25;
    
  Job job = new Job(conf);
  job.setJarByClass(PersonaScoreDriver.class);
  job.setJobName("Persona Score");
  
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(Text.class);

  FileInputFormat.setInputPaths(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
  job.setInputFormatClass(KeyValueTextInputFormat.class);
  
  job.setMapperClass(MyMapper.class);
  job.setNumReduceTasks(0);

  boolean success = job.waitForCompletion(true);
  System.exit(success ? 0:1);  
 }
 
 private static class MyMapper extends Mapper{
  @Override
  public void map(Text key, Text value,
    Context context) throws IOException, InterruptedException {
   
   int s = score;
   String[] personaSplit = value.toString().split(" ");
   
   if(personaSplit.length == 4){
    int mScore = Integer.valueOf(personaSplit[3]);
    
    if(mScore >= s)
     context.write(key, value);
   }
  }
 }
}

Forma Correcta

 
public class PersonaScoreDriver {
 
 private static class MyMapper extends Mapper{
  private int score;
  
  @Override
  protected void setup(Context context) throws IOException,
    InterruptedException {
   Configuration conf = context.getConfiguration();
   //Indicas cuál es el id del parámetro a recoger y el valor
   // por defecto.
   score = conf.getInt("score", 0);
  }
  
  @Override
  public void map(Text key, Text value,
    Context context) throws IOException, InterruptedException {
   
   int s = score;
   String[] personaSplit = value.toString().split(" ");
   
   if(personaSplit.length == 4){
    int mScore = Integer.valueOf(personaSplit[3]);
    
    if(mScore >= s)
     context.write(key, value);
   }
  }
 }
 
 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  //Indicamos el parámetro a pasar
  conf.setInt("score", 25);
    
  Job job = new Job(conf);
  job.setJarByClass(PersonaScoreDriver.class);
  job.setJobName("Persona Score");
  
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(Text.class);

  FileInputFormat.setInputPaths(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
  job.setInputFormatClass(KeyValueTextInputFormat.class);
  
  job.setMapperClass(MyMapper.class);
  job.setNumReduceTasks(0);

  boolean success = job.waitForCompletion(true);
  System.exit(success ? 0:1);  
 }
}

Hay que tener en cuenta que esta forma de pasar información entre el Driver y las tareas (map o reduce) es muy útil siempre cuando esta información no ocupa demasiada memoria ya que esta información se copia en varios sitios: en la memoria del Driver, en la memoria del TaskTracker y en la memoria de la JVM que ejecuta la tarea.
Por lo tanto, si se trata de una información grande (por ejemplo, varios TB), conviene usar otro método (por ejemplo: almacenar esta información en un fichero HDFS y acceder a esta información vía el método setup(), o bien usar Distributed Cache).

Recordad que este código también lo podréis descargar desde este enlace.

Ejemplo de Partitioner

En la entrada anterior vimos qué es el Partitioner, ahora toca ver un ejemplo y el código desarrollado para hacer ese ejemplo.

Atención si vais a hacer las pruebas en local ya que no funciona, el modo local sólo tiene un Reducer y por eso es mejor usar por lo menos el modo pseudo distribuido.

En este ejemplo, a partir del fichero scores.txt con la forma:

01-11-2012 Pepe Perez Gonzalez 21
01-11-2012 Ana Lopez Fernandez 14
15-02-2013 Angel Martin Hernandez 3
01-11-2012 Maria Garcia Martinez 11
01-11-2012 Pablo Sanchez Rodriguez 9
01-11-2012 Angel Martin Hernandez 3
15-01-2013 Pepe Perez Gonzalez 17
15-01-2013 Maria Garcia Martinez 3
...

Queremos dividir los datos por año y enviar a cada Reducer las personas que han jugado en ese año. A través de un Partitioner vamos a indicar a qué Reducer va cada registro.

El Driver es parecido a todo lo que hemos visto hasta ahora, lo único que hemos definido que el InputFormat será un KeyValueTextInputFormat (ya que la fecha está separado por una tabulación del resto de la línea, así que este formato reconocerá la entrada).
Y luego añadimos en las configuraciones nuestra clase Partitioner y el número de tareas Reduce que queremos (nuestro fichero sólo tiene datos de 2 años (2012 y 2013), entonces serán 2 tareas Reducer).

public class PersonaScoreDriver {
 public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  Job job = new Job(conf);
  job.setJarByClass(PersonaScoreDriver.class);
  
  job.setJobName("Persona Score");
  
  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(Text.class);

  FileInputFormat.setInputPaths(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  
  job.setInputFormatClass(KeyValueTextInputFormat.class);
  //Establecemos el número de tareas Reduce
  job.setNumReduceTasks(2);
  
  job.setMapperClass(PersonaScoreMapper.class);
  job.setReducerClass(PersonaScoreReducer.class);
  //Indicamos cuál es nuestro partitioner
  job.setPartitionerClass(PersonaScorePartitioner.class);

  boolean success = job.waitForCompletion(true);
  System.exit(success ? 0:1);  
 }
}

En el Mapper lo único que hacemos es sacar el nombre y apellidos del value y emite un par key/value enviando como key la fecha y como value la persona.

public class PersonaScoreMapper extends 
 Mapper<Text, Text, Text, Text> {
 
 Text persona = new Text();
 
 @Override
 public void map(Text key, Text values,
   Context context) throws IOException, InterruptedException {
  
  String[] personaSplit = values.toString().split(" ");
  StringBuilder persBuilder = new StringBuilder();
  // Puede haber personas con un apellido o con dos
  if(personaSplit.length == 3 || personaSplit.length == 4){
   if(personaSplit.length == 3){
    persBuilder.append(personaSplit[0]).append(" ")
     .append(personaSplit[1]);
   }else {
    persBuilder.append(personaSplit[0]).append(" ")
     .append(personaSplit[1]).append(" ")
     .append(personaSplit[2]);
   }
   persona.set(persBuilder.toString());
   context.write(key, persona);
  }
 }
}

La clase Reducer lo único que hace es recoger la key con su lista de values correspondientes, recorrer esa lista y emitir cada par key/value con la fecha y el nombre. Al haber realizado el Partiiioner, un mismo reducer procesará las keys de un mismo año.

public class PersonaScoreReducer extends 
 Reducer<Text, Text, Text, Text> {

 @Override
 public void reduce(Text key, Iterable<Text> values,
   Context context) throws IOException, InterruptedException {
  for (Text value : values) {
   context.write(key, value);
  }  
 }
}

Por último el Partitioner, que lo que hace es devolver un entero indicando cuál es el Reducer al que irán los datos intermedios generados por el Mapper.

public class PersonaScorePartitioner extends Partitioner<Text, Text> {
 
 @Override
 public int getPartition(Text key, Text value, int numPartitions) {
  
  if(key.toString().endsWith("2012")){
   return 0;
  }else{
   return 1;
  }
 }
}

Una vez visto el desarrollo y cómo quedaría el código sólo quedaría exportar las clases como jar (tal y como vimos en esta entrada) al directorio que tengamos preparado para la ejecución de Jobs en modo pseudo-distribuído y lo lanzaríamos (previamente habiendo puesto en HDFS el fichero scorePartMezcla).
También os recuerdo que las clases las podréis encontrar en la sección de Código Fuente

hadoop jar training/jars/EjemploPartitioner.jar PersonaScoreDriver pruebas/scorePartMezcla pruebas/resultados/ejemploPartitioner

Y este debería ser el resultado si listamos el contenido del directorio ejemploPartitioner:

elena:hadoop elena$ hadoop fs -ls pruebas/resultados/ejemploPartitioner
Found 4 items
-rw-r--r--   1 elena supergroup          0 2013-04-02 19:34 /user/elena/pruebas/resultados/ejemploPartitioner/_SUCCESS
drwxr-xr-x   - elena supergroup          0 2013-04-02 19:34 /user/elena/pruebas/resultados/ejemploPartitioner/_logs
-rw-r--r--   1 elena supergroup        562 2013-04-02 19:34 /user/elena/pruebas/resultados/ejemploPartitioner/part-r-00000
-rw-r--r--   1 elena supergroup        684 2013-04-02 19:34 /user/elena/pruebas/resultados/ejemploPartitioner/part-r-00001


Y en cada fichero quedaría el siguiente contenido.

En el part-r-00000 que correspondería al año 2012 y que le habíamos asignado el valor 0:


elena:hadoop elena$ hadoop-1.0.4/bin/hadoop fs -cat pruebas/resultados/ejemploPartitioner/part-r-00000
01-11-2012 Angel Martin Hernandez
01-11-2012 Maria Garcia Martinez
01-11-2012 Ana Lopez Fernandez
01-11-2012 Pablo Sanchez Rodriguez
01-11-2012 Pepe Perez Gonzalez
01-12-2012 Maria Garcia Martinez
01-12-2012 Pepe Perez Gonzalez
01-12-2012 Pablo Sanchez Rodriguez
01-12-2012 Ana Lopez Fernandez
15-11-2012 Pepe Perez Gonzalez
15-11-2012 Maria Garcia Martinez
15-11-2012 John Smith
15-11-2012 Cristina Ruiz Gomez
15-12-2012 John Smith
15-12-2012 Cristina Ruiz Gomez
15-12-2012 Maria Garcia Martinez
15-12-2012 Pepe Perez Gonzalez
15-12-2012 Angel Martin Hernandez


En el part-r-00001 que correspondería al año 2013 y que le habíamos asignado el valor 1:

elena:hadoop elena$ hadoop-1.0.4/bin/hadoop fs -cat pruebas/resultados/ejemploPartitioner/part-r-00001
01-01-2013 Ana Lopez Fernandez
01-01-2013 John Smith
01-01-2013 Pablo Sanchez Rodriguez
01-01-2013 Pepe Perez Gonzalez
01-01-2013 Maria Garcia Martinez
01-01-2013 Angel Martin Hernandez
01-02-2013 Ana Lopez Fernandez
01-02-2013 Cristina Ruiz Gomez
01-02-2013 Maria Garcia Martinez
01-02-2013 Pepe Perez Gonzalez
15-01-2013 Angel Martin Hernandez
15-01-2013 Maria Garcia Martinez
15-01-2013 Pepe Perez Gonzalez
15-01-2013 John Smith
15-01-2013 Pablo Sanchez Rodriguez
15-02-2013 Pepe Perez Gonzalez
15-02-2013 John Smith
15-02-2013 Pablo Sanchez Rodriguez
15-02-2013 Maria Garcia Martinez
15-02-2013 Ana Lopez Fernandez
15-02-2013 Cristina Ruiz Gomez
15-02-2013 Angel Martin Hernandez