MapReduce, le framework de traitement de données de Hadoop
Avec HDFS et YARN, MapReduce constitue le troisième pilier du noyau de Hadoop. Il s'agit d'un modèle de traitement de données inventé par Google en 2004 pour réduire les temps de traitement des données en grande quantité en utilisant le système de fichier distribué HDFS. MapReduce organise le traitement des données en plusieurs petites tâches qui peuvent être exécutées parallèllement puis agrégées afin de produire le résultat escompté.
Dans cet article, nous allons en découvrir plus sur son architecture et son fonctionnement mais aussi écrire notre premier job MapReduce en Python 🤓.
Architecture de MapReduce
Chaque algorithme de traitement de données basé sur MapReduce ou encore job mapReduce doit contenir deux types de programmes :
- Les Mappers : Ce sont des fonctions qui reçoivent en entrée les données à traiter et produisent pour chaque ligne de données, un résultat sous forme d'une paire (clé,valeur).
- Les Reducers : Ces fonctions reçoivent les résultats des mappers et les réduisent au moyen de diverses sortes d'opérations afin d'en obtenir un résultat plus concis correspondant à celui attendu par l'utilisateur.
Voici en image, un aperçu des étapes d'un job MapReduce :
Supposons que vous ayez vos données stockées dans un cluster HDFS sur lequel vous exécutez un job MapReduce. L’exécution d’un programme se déroulera selon les étapes suivantes :
- La phase d'input : Les données HDFS sont lues et transmisses bloc par bloc chacune à un processus map séparé qui travaille uniquement sur les données qui lui sont assignées.
- La phase Map : Pendant cette phase, les instances de notre fonction map (que nous avons définie) s'exécute sur les blocs de données parallèllement et produit 0 ou plusieurs paires de (clé, valeur) appelées clés intermédiaires. Nous verrons un peu plus bas à quoi correspondent ces clés, ces valeurs mais aussi le contenu d'un programme MapReduce.
- La phase combiner (optionnelle) : Écrire une fonction Combiner n'est pas idispensable dans un programme MapReduce mais elle est utilisée pour réduire la taille des résultats du Mapper avant qu'ils soient transmis aux reducers.
- La phase Shuffle and Sort : Cette étape est automatiquement réalisée par MapReduce sans besoin qu'une fonction soit définie par l'utilisateur. Ici, MapReduce regroupe ensemble les résultats des différents mappers puis les trie par ordre croissant de leurs clés avant de les assigner, groupe par groupe à différents reducers.
- La phase Reducer : Un reducer est une fonction définie par l'utilisateur (tout comme un mappr) qui combine les groupes de paires (clé,valeur) en utilisant plusieurs techniques de filtrage,de sommage ou tout autre type d'opérations de groupage. Le reducer renvoie un set généralement très concis de paires (clé,valeur) à la dernière phase.
- La phase d'Output : Dans cette phase, les résultats du reducer sont lus et affichés à l'utilisateur ou (plus fréquemment) écrits dans un fichier.
Fonctionnement de MapReduce
Le fonctionnemment de MapReduce est intimement lié à l'architecture de fonctionnement de HDFS.
Le fonctionnement d'un job MapReduce suit les étapes ci-dessous :
- Étape 1 : Le client soumet un job MapReduce (Un programme complet de traitement de données). Un exemple de job est par exemple, un programme qui renvoie la somme des ventes d'un produit pendant un trimestre. Dans cette première phase, le code du job MapReduce (mappers, reducers et combiners) est soumis au gestionnaire de ressources YARN et les données dont le job a besoin sont copiées sur HDFS simultanément (au cas où elles n’y sont pas déjà).
- Étape 2 : YARN détermine les ressources nécessaires à l'exécution du job et effectue la répartition des tâches aux différents serveurs du cluster. Cette répartition prend en compte la disponibilité de ces serveurs mais aussi la localisation des données (l'objectif étant que le code doit être le plus proche possible des données)
- Étape 3 : L’Application Master lance, coordonen et surveille les tâches map et reduce sur les autres noeuds du cluster. L’Application Master est similaire à la fonction main d’un programme orchestrant l'exécution de toutes les autres fonctions/méthodes de la classe.
- Étape 4 : Les serveurs esclaves exécutent les tâches MapReduce en utilisant les données dans HDFS. Il faut noter que le Node Manager est un agent de YARN qui surveille l'utilisation mémoire et processeur de chaque noeud ainsi que sa performance pour la communiquer à YARN. Il permet en cas de défaillance d'un noeud, d'alerter YARN pour que les tâches interrompues puissent être relancées sur un autre serveur.
PS : Les notions de JobTracker et TaskTracker existant avec MapReduce 1 ont été remplacées respectivement par l'Application Master et le NodeManager avec l'avenue de MapReduce 2.
PS2 : En cas de défaillance de l'Application Master, YARN le redémarre et en cas de défaillance de YARN lui-même, le mode high availability pour haute disponibité peut être configuré pour permettre à un gestionnaire de configuration distribué comme Zookeeper de redémarrer un nouveau gestionnaire de ressources.
Au code moussaillons ! 💪🏾
Tout ce verbillage technique bien que nécessaire peut paraître intimidant pour un programmeur non averti. Pour faire simple, un job MapReduce est défini par une classe et les mappers, les reducers et les combiners par des fonctions de cette classe.
MapReduce est écrit en Java et donc les jobs MapReduces sont normalement écrits en Java. Cependant grâce à l'API Hadoop Streaming, il est maintenant possible d'écrire des jobs MapReduce dans plusieurs langages dont Python, PHP, Perl ou encore C++.
Bien que les jobs MapReduce écrits avec des langages autres que Java soient plus lents (ils doivent évidemment être convertis en bytecode Java avant d'être exécutés), ces alternatives demeurent une excellente porte d'entrée au monde de MapReduce pour les non-javaistes.
Cas d'utilisation
Pour notre premier job MapReduce, nous allons réaliser un petit exercice en utilisant le fameux dataset de IMDB de films et de divertissements audio-visuels de tout genre. Allons-y!
- Acquisition des données : Téléchargez et décompressez les données que nous utiliserons avec les commandes suivantes :
$ wget https://datasets.imdbws.com/title.basics.tsv.gz$ gunzip title.basics.tsv.gz#Cette commande vous permet de voir combien de lignes contient le fichier
$ wc -l title.ratings.tsv
La commande wc -l qui permet d'afficher le nombre de lignes d'un fichier, montre ici que le dataset contient plusieurs millions de lignes. Nous allons donc extraire les 100 000 dernières lignes pour notre travail.
NB : MapReduce permet d'exploiter les clusters pour un traitement rapide; il ne décuple pas les capacités de votre machine .Vous pouvez donc essayer de travailler avec le dataset complet aux risques et périls de votre machine 😅.
On continue en explorant notre fichier pour en découvrir la structure avant d'en extraire les 100 000 lignes qui nous intéressent :
#Cette commande nous donne un aperçu des données
$ head title.ratings.tsv#Les 100 000 dernières lignes sont transférées dans un nouveau fichier
$ tail -n 100000 title.basics.tsv > moviesSubset.tsv
Voici ce à quoi ressemble notre fichier. Nous pouvons observer différents attributs du type d'élément au titre du film, au genre en passant par l'année de sortie.
Pour notre exercice, la question à laquelle, nous allons répondre est la suivante :
Combien d'éléments de chaque type contient le dataset ?
Voici à quoi ressemble le job MapReduce qui va nous aider à répondre à cette question :
Prenez le temps de scanner ce petit extrait de code attentivement. Si vous avez déjà fait de la programmation python, ce code ne devrait rien avoir de sorcier pour vous.
- La classe correspond au job mapReduce qui est exécuté en fin de script.
- Les mappers et reducers (un seul de chaque dans ce cas) correspondent aux fonctions de la classe.
Dans notre exercice, le mapper convertit chaque ligne du fichier en tuple avant de retourner le type d'entrée sous forme de paire (clé,valeur) ou le type d'entrée (short, movie, tvseries etc) est la clé et "1" la valeur.
Le réducer groupe ensuite toutes ces paires (clé, valeur) en sommant toutes les valeurs correspondant à une même clé. Le reducer retourne donc au final, une liste de clés (types d'entrée) associée chacune avec le nombre d'entrées qui y correspondent dans le dataset.
- Le(s) mapper(s) et reducer(s) sont ordonnés dans une fonction step qui construit séquentiellement les différentes étapes du job MapReduce qui doivent être exécutées l'une après l'autre.
Pour exécuter ce job, la commande est la suivante :
$ python3 NombreDeFilmsParTypes.py -r hadoop — hadoop-streaming-jar /home/hdoop/hadoop-3.2.1/share//hadoop/tools/lib/hadoop-streaming-3.2.1.jar moviesSubset.tsvLa localisation de votre hadoop-streaming-jar sera différente selon votre type d’installation Hadoop. Une fois le job lancé, MapReduce copie les données sur HDFS puis fait appel au gestionnaire de ressources (voir image suivante).
Ensuite, après quelques secondes (ou minutes selon les capacités de votre machine), vous obtenez des résultats similaires à ceux montrés ci-dessous. Le dataset contient donc 82440 épisodes de séries, 3853 films et 230 jeux vidéos entre autres.
Exercice d'application
Cet article était une brève introduction pratique à MapReduce. Pour aller plus loin, la documentation de MrJob est votre meilleur ami. En attendant, vous pouvez vous amuser, en utilisant le même dataset, à essayer d'écrire un job MapReduce pour obtenir les années par ordre croissant du nombre total de vidéos/films sortis .
Bonne chance ! :-)
PS : Yo ! Si ce contenu vous a été utile d'une quelconque manière que ce soit, n'hésitez pas à me fournir du carburant pour en écrire d'autres en me laissant des 👏🏾. Plus d'articles sur Pig, Hive et Spark arrivent bientôt.
En attendant, jetez un coup d'oeil à mes autres articles introduisant le Big Data, ses technologies et enjeux, sur l'installation de Hadoop sous Ubuntu, et la prise en main de HDFS, le système de fichier star de Hadoop.
À très vite!
