Map Reduce Patterns 2

Le but de ce TP est d'implémenter un pattern courant en Map Reduce : le "join".

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/$

Partie 1 : Reduce-side join

Le but de cette partie est d'implémenter un inner join dans le reducer. Le inner join ne conserve que les éléments qui sont présents dans les deux jeux de données.

On effectura le join entre les jeux de données worldcitiespop.txt et region_codes.csv (qui contient le nom de chaque région en fonction de son identifiant pays et de l'identifiant de la région).

Exercice 1 : TaggedValue

Ecrire une classe TaggedValue pour avoir un type unique pour les villes et les régions. Elle doit comporter un champ permettant de différencier les deux et un champ contenant la valeur correspondante sous forme de texte. Cette classe servira de valeur pour le job Map Reduce, elle doit donc implémenter l'interface Writable.

Exercice 2 : Mappers

Ecrire deux mappers différents pour parser les deux jeux de données. La sortie des deux mappers doit etre du type TaggedValue. On utilisera ensuite la classe org.apache.hadoop.mapreduce.lib.input.MultipleInputs et sa methode addInputPath lors de la configuration du job pour associer chaque mapper à son jeu de données.

Exercice 3 : Reducer

Ecrire le reducer pour effectuer le join. La sortie doit comporter le nom de chaque ville avec le nom de sa région, sous forme de texte.

Attention : Les valeurs ne sont pas triées en entrée du reducer. La région n'est donc pas forcément la première valeur. De plus, si le reducer ne trouve pas de région ou plusieurs régions, c'est qu'il y a un problème.

Exercice 4 : Job

Configurer et tester le job.

Partie 2 : Scalable join

Vous pouvez remarquer qu'avec l'implémentation de la première partie, dans le pire des cas, toutes les valeurs pour une clé doivent être lues en mémoire avant de pouvoir effectuer le join. Cette solution pose des problèmes evidents dans le cas ou le nombre de valeurs devient important. Dans cette partie, on ajoutera du code au reduce-side join pour pallier ce problème. L'objectif principal est de faire en sorte que les valeurs arrivent triées dans le reducer. De cette manière, toutes les valeurs du premier dataset sont avant celles du second dataset et seules les premières valeurs doivent être lues en mémoire.

Exercice 1 : TaggedKey

Ecrire une classe TaggedKey, qui contient la clé utilisée précédemment (apellée clé naturelle), ainsi que l'identifiant du jeu de données. Comme cette classe doit pouvoir servir de clé, elle doit implémenter WritableComparable.

Exercice 2 : Partitionner

Maintenant que la clé contient aussi l'identifiant que dataset, le partitionnement par défaut d'hadoop ne donne pas le résultat voulu. En effet, on ne veut partitionner qu'en fonction de la clé naturelle. Pour cela, implémentez un org.apache.hadoop.mapreduce.Partitioner qui assigne les clés aux partitions en fonction du hashcode de la clé naturelle.

Exercice 3 : Grouping & sort

Une fois les clés correctement assignées aux partitions, il reste à les grouper et les trier en entrée du reducer. Pour cela, implémentez deux WritableComparator. Le premier sera utilisé pour le grouping et fera en sorte que toute les clés qui ont la même clé naturelle soit groupées (c.à.d. égales). Le second servira à trier les clés, en plus de la comparaison par clé naturelle, il doit s'assurer que les regions arrivent avant les villes.

Exercice 4 : Job

Utilisez les méthdes de Job setPartitionerClass, setGroupingComparatorClass et setSortComparatorClass pour indiquer à hadoop les classes à utiliser. Modifiez le reducer pour prendre en compte le nouvel ordre d'arrivée des valeurs et testez le job.