Premiers pas avec Hadoop MapReduce 2/2

Le but de ce TP est de continuer à se familiariser avec le framework Map Reduce implémenté dans l'écosystème Hadoop. Pour ce TP on considère que nous disposons d'un serveur Hadoop YARN. On installera (si-nécessaire) uniquement les outils permettant de communiquer avec ce serveur YARN (Hadoop ressource manager). Addresse des services du cluster:

  • Namenode (HDFS): $hdfs://10.0.NUM_SALLE.3:9000/$
  • Ressource manager (YARN): $10.0.NUM_SALLE.3:$
  • WebUI (HDFS): $http://10.0.NUM_SALLE.3:50070/$
  • WebUI (YARN): $http://10.0.NUM_SALLE.3:8088/$

RAPPEL : Squelette minimal d'une application Map Reduce

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritbale;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextOutputFormat;

public class MonProg {
  public static class MonProgMapper
       extends Mapper<LongWritable, Text, Text, IntWritable>{
      public void map(Object key, Text value, Context context
              ) throws IOException, InterruptedException {
          //context.write(word, one);
      }
  }
  public static class MonProgReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      //context.write(key, result);
    }
  }
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "MonProg");
    job.setNumReduceTasks(1);
    job.setJarByClass(MonProg.class);
    job.setMapperClass(MonProgMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setReducerClass(MonProgReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    job.setInputFormatClass(TextInputFormat.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

Archive du projet maven : mapreduce_maven.tgz

Compilation

Compiler avec Maven : mvn compile

Générer le jar sans Maven :

cd mapreduce_maven/target/classes
jar cf tp3.jar *.class

Attention, pensez bien à spécifier la main class lors de l'appel à yarn dans ce cas.

Générer le jar avec Maven : mvn package

Attention, pensez il ne faut pas spécifier la main class lors de l'appel à yarn dans ce cas. Car la main est déjà spécifiée dans le "manifest"

Données

Dans le répertoire /user/raw_data de HDFS, vous trouverez un fichier nommé worldcitiespop.txt de taille 144MB. Ce fichier contient la liste des villes de la planète, leur géolocalisation, leur code région et leur population.

Exercie 1: Histogramme

Ecrivez un programme MapReduce qui calcul l'histogramme de fréquence des villes dont ont a la population. On utilisera une échelle logarithmique pour déterminer les classes d'équivalences des villes. Soit v1 et v2 deux villes v1 et v2 sont dans la même clasee d'équivalence si et seulement si int(log(pop(v1))) == int(log(pop(v_2))).

Le fichier généré par votre "reducer" devrait ressembler à cela:

10       45
100  585
1000     9429
10000    27065
100000   9820
1000000  979
10000000 57

Dans ce tableau on peut voir qu'il y a 45 villes avec entre 0 et 10 habitants et 27065 villes avec 10000 habitants. Visualisez votre résultat à l'aide d'un tableur.

Exercie 2: Résumé

Modifiez votre programme précédent pour ajouter le calcul de la population moyenne, de la population minimale et de la population maximale de chaque classe d'équilavence.

Le fichier généré par votre "reducer" devrait ressembler à cela:

classe   count    avg    max   min
10       45       xx     xx     xx
100  585      xx     xx     xx
1000     9429     xx     xx     xx
10000    27065    xx     xx     xx
100000   9820     xx     xx     xx
1000000  979      xx     xx     xx
10000000 57       xx     xx     xx

Exercice 3: Paramètrage

En utilisant la fonction getConfiguration() accessible dans les classes Mapper et Reducer et en redéfinissant la fonction setup de vos mapper/reducer. Modifiez le code de l'exercice précédent pour permettre de paramétrer le code de génération de votre histogramme. On pourra ainsi avoir une discrétisation plus ou moins précise en fonction d'un paramètre passé sur la ligne de commande de votre programme.