Recuerda que puedes descargarte algunos de los ejemplos en la pestaña de Código Fuente

viernes, 3 de mayo de 2013

ToolRunner

Cuando desarrollamos una aplicación Hadoop, quizás necesitemos en algún momento utilizar opciones de la línea de comandos Hadoop, para lo cual se utiliza la clase ToolRunner.

GenericOptionsParser es quien interpreta esas opciones de línea de comandos de Hadoop.
Normalmente no se usa esta clase, sino que se usa ToolRunner que la utiliza internamente y se suele usar en el Driver de un programa MapReduce

Así que si en algún momento queremos pasar opciones de configuración a través de la línea de comandos, en nuestro programa deberemos utilizar ToolRunner e invocando al método ToolRunner.run().

Para pasar opciones de configuración a través de línea de comando se usaría a través de la opción:
-D propiedad=valor

Supongamos que lanzamos la ejecución de un programa MapReduce a través de la línea de comandos de esta forma:

hadoop jar WordCount.jar WordCountDriver -D mapreduce.job.reduces=5 -D mapreduce.job.maps=10 directorioInput directorioOutput


Si no usáramos el ToolRunner tendríamos que controlar las propiedades introducidas haciendo algo parecido a esto:

public static void main(String[] args) throws Exception {

  Configuration conf = new Configuration();
  ...
  conf.set("prop1",args[0]);  
  conf.set("prop2", args[1]);
  ...
  Job job = new Job(conf);
  ...
  FileInputFormat.setInputPaths(job, new Path(args[2]));
  FileOutputFormat.setOutputPath(job, new Path(args[3]));
  ...
}


Mientras que si utilizamos el ToolRunner, la función run se encarga de parsear por si sólo los argumentos de configuración que hayamos introducido, siempre y cuando hayamos puesto -D delante de la propiedad.
Para desarrollar a través del ToolRunner la clase debe extender de Configured e implementar Tools. Resultando de la forma:


public class WordCountDriver extends Configured implements Tool{
 public int run(String[] args) throws Exception {
  Configuration conf = new Configuration();
  Job job = new Job(conf);
  ...
  FileInputFormat.setInputPaths(job, new Path(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
   ...
 }

 public static void main(String[] args) throws Exception {
       int exitCode = ToolRunner.run(new Configuration(), 
            new WordCountDriver(), args);
       System.exit(exitCode);
 }
}


Si os preguntáis cuál es la diferencia y las ventajas entre usar ToolRunner o llamar directamente a la función main con todas las configuraciones, realmente el funcionamiento es el mismo, pero utilizando el ToolRunner es mucho más limpio y no usarlo puede llevarnos a errores como asignar las propiedades después de haber instanciado el Job (que esa propiedad sería nula).

public static void main(String[] args) throws Exception {

  Configuration conf = new Configuration();
  Job job = new Job(conf);
  ...
  conf.set("prop1",args[0]);  
  ...
  // Aquí "prop1 sería nulo en el job
  job.getConfiguration().get("prop1"); 
}


Además de la opción -D que nos permite pasar parámetros a la ejecución, hay que saber que existen otras opciones de línea de comandos que nos permiten:
-files fich1, fich2, ... : Que copia los ficheros indicados del sistema de ficheros local al sistema de ficheros HDFS para utilizarlos en el programa MapReduce.
-libjars jar1, jar2, ... : Que copia las librerías jar especificadas del sistema de ficheros local a HDFS y lo añade al Classpath de la tarea MapReduce (muy útil para añadir dependencias Jar de los Job).
-archives archive1, archive2, ... : Que copia los archivos del sistema de ficheros local a HDFS y poniéndolos a disposición de los programas MapReduce.




domingo, 28 de abril de 2013

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




lunes, 22 de abril de 2013

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.