TP1 - Map-Reduce sur Hadoop

Mise en route

Pour lancer la machine virtuelle sur laquelle est installé hadoop, lancez la commande suivante depuis un terminal.

/users/usrlocal/bin/kvm-mbds

Une fois la machine virtuelle lancée utilisez le login mbds et mot de passe mbds pour vous connecter et lancer la commande startx pour démarrer le serveur X (minimaliste). Un clic droit sur le bureau vous permet d'accéder aux différentes applications, pour ce TP on aura besoin du terminal, du navigateur web et de geany pour écrire les fichiers pythons.

Rappels Unix ("Le terminal, c'est pas facile...")

Mais si... Quelques rappels en vrac :

cd dirse déplacer vers le répertoire dir (qui doit exister dans le répertoire courant).
cd ..se déplacer dans le répertoire parent.
mkdir dircréer le répertoire dir dans le répertoire courant.
pwd afficher le répertoire courant.
ls afficher le contenu du répertoire courant.
python script.py   exécuter le fichier python script.py (qui doit exister dans le répertoire courant !).
rm script.py supprimer le fichier script.py.
rm -R dir supprimer le répertoire dir.
echo "bonjour" Afficher bonjour à l'écran.
cat file Afficher le contenu du fichier file à l'écran.

Créez un répertoire TP1 dans le répertoire utilisateur de la machine virtuelle et un sous-répertoire comptage dans TP1, déplacez vous dans ce répertoire pour la suite du TP.

Utilisation de HDFS

HDFS est le sytème de fichiers d'hadoop qui permet de manipuler de manière transparente des données réparties sur un grand nombre de noeuds. Pour ce TP nous n'utiliserons pas de gros volumes de donnée et ces données seront stockées sur un unique noeud, mais gardez à l'esprit que les programmes et manipulations de ce TP seraient les mêmes si on avait affaire à de grandes quantités de données réparties sur plusieurs noeuds.

Utiliser HDFS est très similaire à l'utilisation des commandes unix habituelles, le schéma de base est le suivant :

hadoop fs -cmd [args]

cmd est une commande unix ou une commande spécifique à hdfs (voir la doc pour une liste des commandes disponibles).

Quelques exemples :

hadoop fs -mkdir /dircréer le répertoire dir à la racine du système de fichier
hadoop fs -ls /dir afficher le contenu du répertoire /dir.
hadoop fs -copyFromLocal ./data.txt /dir    copier le fichier data.txt du répertorie courant dans le répertoire /dir sur le système de fichier d'hadoop (on a copié le fichier de la machine locale vers le noeud distant).

Dans un terminal, lancez les deux scripts start-dfs.sh puis start-yarn.sh pour démarrer le serveur hadoop. Exemple de sortie :

mbds@mbds:~/TP1/comptage$ start-dfs.sh
Starting namenodes on [localhost]
localhost: starting namenode, logging to /usr/local/hadoop/logs/hadoop-mbds-namenode-mbds.out
localhost: starting datanode, logging to /usr/local/hadoop/logs/hadoop-mbds-datanode-mbds.out
Starting secondary namenodes [0.0.0.0]
0.0.0.0: starting secondarynamenode, logging to /usr/local/hadoop/logs/hadoop-mbds-secondarynamenode-mbds.out
mbds@mbds:~/TP1/comptage$ start-yarn.sh
starting yarn daemons
starting resourcemanager, logging to /usr/local/hadoop/logs/yarn-mbds-resourcemanager-mbds.out
localhost: starting nodemanager, logging to /usr/local/hadoop/logs/yarn-mbds-nodemanager-mbds.out

Exécutez ensuite les commandes suivantes pour vérifier que tout fonctionne (normalement vous devez comprendre ce que font ces commandes !) :

mbds@mbds:~/TP1/comptage$ hadoop fs -mkdir /test
mbds@mbds:~/TP1/comptage$ hadoop fs -ls /
Found 1 items
drwxr-xr-x - mbds supergroup 0 2016-01-27 16:28 /test
mbds@mbds:~/TP1/comptage$ hadoop fs -rm -R /test
16/01/27 16:28:35 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 0 minutes, Emptier interval = 0 minutes.
Deleted /test
mbds@mbds:~/TP1/comptage$ hadoop fs -ls /

Premier programme Map-Reduce : comptage

Pour la première partie de ce TP nous allons implémenter le hello world du monde Map-Reduce/Hadoop : le comptage de mots.

Récupérer les données

Nous allons utiliser des livres électroniques accessibles sur le projet Gutenberg. Téléchargez les 3 livres suivants (au format texte, lien Plain Text UTF-8) :

Créez ensuite un répertoire gutenberg à la racine du système de fichier d'hadoop et copiez-y les 3 fichiers textes (utilisez la commande -copyFromLocal). Pour vérifier que tout c'est bien passé afficher le contenu du répertoire /gutenberg :

mbds@mbds:~/TP1/comptage$ hadoop fs -ls /gutenberg
Found 3 items
-rw-r--r-- 1 mbds supergroup 1428841 2016-01-27 20:42 /gutenberg/5000-8.txt
-rw-r--r-- 1 mbds supergroup 674570 2016-01-27 20:42 /gutenberg/pg20417.txt
-rw-r--r-- 1 mbds supergroup 1573151 2016-01-27 20:42 /gutenberg/pg4300.txt

Map

Nous allons écrire le mapper pour notre tâche de comptage. Le rôle du mapper est d'émettre le couple (mot,1) pour chaque mot du document qui lui est donné en entrée. Nous allons écrire ce mapper en python, le document sera donné via l'entrée standard et l'émission du couple (clé,valeur) consistera à afficher

clé valeur

à l'écran.

Quelques rappels et infos utiles sur python :

print "bonjour"affiche bonjour à l'écran (i.e. écrit bonjour sur la sortie standard.
s = "Longtemps je me suis"
L = s.split(" ")
la méthode .split(" ") découpe une chaîne de caractère en mots. On obtient donc L = ["Longtemps", "je", "me", "suis"].
line.strip() supprime le retour à la ligne à la fin de la chaîne de caractères line
import sys
for line in sys.stdin:
    print line
affiche chaque ligne de l'entrée standard à l'écran.

Vous aves maintenant tous les outils pour écrire le mapper. Utilisez geany (clic droit sur le bureau -> Debian -> Applications -> Programming) pour écrire le mapper en python que vous sauvegarderez sous le nom mapper.py dans le répertoire comptage. Utilisez les commandes suivantes pour vérifier que votre mapper fonctionne correctement :

mbds@mbds:~/TP1/comptage$ echo "longtemps je me suis je me suis je me" | python mapper.py
longtemps 1
je 1
me 1
suis 1
je 1
me 1
suis 1
je 1
me 1
mbds@mbds:~/TP1/comptage$ echo "" | python mapper.py
mbds@mbds:~/TP1/comptage$ echo " " | python mapper.py

Shuffle

L'étape de shuffle consiste à rassembler les couples (clé,valeur) par clés, pour nos tests il suffira de trier la sortie du mapper en utilisant la commande unix :

mbds@mbds:~/TP1/comptage$ echo "longtemps je me suis je me suis je me" | python mapper.py | sort
je 1
je 1
je 1
longtemps 1
me 1
me 1
me 1
suis 1
suis 1
Cette phase de map/reduce era gérée par hadoop, donc pas de code à écrire pour le shuffler.

Reduce

Le rôle du reducer pour la tache de comptage est de sommer les valeurs correspondant à une même clé est d'émettre le couple (clé,somme). L'entrée donnée au reducer est une liste de (clé,valeur) triée par clés. Chaque ligne de l'entrée contiendra un couple clé valeur séparé par un espace comme dans l'exemple ci dessus. Pour cet exemple le reducer devra afficher :

mbds@mbds:~/TP1/comptage$ echo "longtemps je me suis je me suis je me" | python mapper.py | sort
je 3
longtemps 1
me 3
suis 2
Ici encore l'entrée sera donné via l'entrée standard et l'émission du couple (clé,valeur) consistera à l'afficher à l'écran (i.e. l'écrire sur la sortie standard).

Ecrivez le reducer en python, sauvegardez votre programme sous le nom reducer.py et tester le avec les commandes suivantes :

mbds@mbds:~/TP1/comptage$ echo "longtemps je me suis je me suis je me" | python mapper.py | sort | python reducer.py
je 3
longtemps 1
me 3
suis 2
mbds@mbds:~/TP1/comptage$ cat /usr/share/doc/python-matplotlib/README.txt | python mapper.py | sort | python reducer.py
: 2
:: 1
* 2
*** 6
# 3
10.3.9 1
2.5 1
2.6 1
2D 1
8.4.x 1
a 5
across 1
Agg 1
(ala 1
all 1
also 1
and 6
any 1
Apple's 1

...

which 1
wish 2
with 1
WXAgg 3
wxPython 1
X 1
X. 1
you 5
You 1
your 1

Avant de passer à la suite (et si ce n'est pas déjà fait) ajoutez la ligne suivante au début des fichiers mapper.py et reducer.py

#!/usr/bin/env python

Cette ligne permet à hadoop de savoir que votre programme est écrit en python.

Big Data nous voila...

Maintenant que notre mapper et notre reducer sont fin prêts, on va pouvoir lancer notre tâche colossale de comptage sur hadoop ! Cela se fait avec la commande suivante :

hadoop jar /usr/local/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar -files mapper.py,reducer.py -mapper mapper.py -reducer reducer.py -input /gutenberg/*.txt -output /result

Après exécution de la tâche de comptage on a quelque chose qui ressemble à :

mbds@mbds:~/TP1/comptage$ hadoop jar /usr/local/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar -files mapper.py,reducer.py -mapper mapper.py -reducer reducer.py -input /gutenberg/*.txt -output /result
16/01/27 22:59:01 INFO Configuration.deprecation: session.id is deprecated. Instead, use dfs.metrics.session-id
16/01/27 22:59:01 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
16/01/27 22:59:01 INFO jvm.JvmMetrics: Cannot initialize JVM Metrics with processName=JobTracker, sessionId= - already initialized
16/01/27 22:59:02 INFO mapred.FileInputFormat: Total input paths to process : 3
16/01/27 22:59:02 INFO mapreduce.JobSubmitter: number of splits:3
16/01/27 22:59:02 INFO Configuration.deprecation: user.name is deprecated. Instead, use mapreduce.job.user.name

...

16/01/27 22:59:24 INFO mapreduce.Job: Counters: 32
File System Counters
FILE: Number of bytes read=7130242
FILE: Number of bytes written=23011317
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=11928267
HDFS: Number of bytes written=966052
HDFS: Number of read operations=49
HDFS: Number of large read operations=0
HDFS: Number of write operations=6
Map-Reduce Framework
Map input records=78578
Map output records=629920
Map output bytes=5453510
Map output materialized bytes=6713368
Input split bytes=283
Combine input records=0
Combine output records=0
Reduce input groups=82618
Reduce shuffle bytes=0
Reduce input records=629920
Reduce output records=82618
Spilled Records=1259840
Shuffled Maps =0
Failed Shuffles=0
Merged Map outputs=0
GC time elapsed (ms)=607
CPU time spent (ms)=0
Physical memory (bytes) snapshot=0
Virtual memory (bytes) snapshot=0
Total committed heap usage (bytes)=786538496
File Input Format Counters
Bytes Read=3676562
File Output Format Counters
Bytes Written=966052

Pour consulter les résultats vous pouvez exécuter les commandes suivantes :

mbds@mbds:~/TP1/comptage$ hadoop fs -ls /result
Found 2 items
-rw-r--r-- 1 mbds supergroup 0 2016-01-27 22:59 /result/_SUCCESS
-rw-r--r-- 1 mbds supergroup 966052 2016-01-27 22:59 /result/part-00000
mbds@mbds:~/TP1/comptage$ hadoop fs -cat /result/part-00000

...
Courtesy 1
Courthouse 1
Courting 1
Courts 1
Cousin 1
Cousin. 1
Cousins, 1
Coventry 1
Covered 1
Covoni,--A 1
Cow 2
Cow, 2
Cowe 2
Cowley 27
Cowley's 5
Cowley, 6
Cowley. 6
Cows 1
Cows, 1
Cox 1
...

Vous pouvez copier ces résultats dans un fichier result.txt dans le répertoire courant en utilisant la commande

hadoop fs -cat /result/part-00000 >result.txt

ce qui vous permet de consulter tout le fichier en utilisant par exemple geany. Notez que dans le cas général ceci n'est pas une bonne idée, si les données sont très volumineuses il ne vaut mieux pas les copier sur le disque local, ni ouvrir ce fichier dans un éditeur de texte !

Nettoyage des données

Comme vous avez pu le constater en regardant le fichier de résultats notre comptage est loin d'être parfait... On a notamment les signes de ponctuations qui font interpréter les mots (Cousin), (Cousin,) et (Cousin.) comme différents, ou le 's possessif de l'anglais qui pose le même problème. On a également qu'un mot contenant une majuscule ne correspondra pas à la même clé que ce mot sans majuscule... Modifiez le code de votre mapper pour tenter de nettoyer le plus possible les clés avant l'émission du couple (clé,valeur).

Loi de Zipf

La loi de Zipf est une observation empirique concernant (entre autres) la fréquence des mots dans un texte. Cette loi énonce grosso modo que le mot le plus fréquent d'un corpus apparait 2 fois plus souvent que le second, 3 fois plus souvent que le troisième, 4 fois plus souvent que le quatrième...

Nous allons vérifier si cette loi s'applique à notre corpus de textes. Ecrivez un programme python qui trie les mots du fichier result.txt en fonction de leur nombre d'occurrences et affichez les 30 mots les plus fréquents avec leur nombre d'occurrences.

Utiliser ensuite le package matplotlib pour dessiner le graphe du nombre d'occurrences des 100 mots les plus fréquents en fonction de leur rang. Si la loi de Zipf est vérifiée ce graphe devrait être proche d'une droite lorsque les deux axes sont en échelle logarithmique.





TP2 - Anagrammes

Anagrammes

La tâche que l'on considère est celle de regrouper des mots en anagrammes. Des anagrammes sont des mots composés des même lettres, comme par exemple chien, chine et niche. Nous utiliserons comme données d'entrée la liste officielle des mots autorisés au Scrabble que vous pouvez téléchargez ici.

Placez ce fichier texte dans un nouveau répertoire sur le système de fichier d'hadoop.

Essayez de trouver par vous même comment résoudre ce problème en utilisant map/reduce. Quelle doit être la sortie finale ? Qu'est ce que deux anagrammes ont en commun ? Que peut-on utiliser comme clé pour un groupe d'anagrammes ? ...

Une fois votre stratégie établie, écrivez le mapper et le reducer en python en vous inspirant des programmes du TP précédent.

Lancez le programme map reduce sur hadoop, récupérez le fichier de résultat et trouvez le groupe d'anagrammes le plus gros.

Envoyez un mail à l'adresse guillaume.rabusseau@lif.univ-mrs.fr avec comme sujet ECM Big Data TP2, mettez le groupe d'anagrammes le plus gros dans le corp du mail et ajouter les fichiers mapper.py et reducer.py en pièce jointe.

Pour les rapides...

Si il vous reste du temps téléchargez cette archive (500 Mo), elle contient les géo-localisations d'articles wikipedia. Ecrivez un programme map reduce qui regroupe les articles wikipedia provenant d'un même endroit.