Chapitre 2 - Hadoop - Universit de Jendouba 2024/2025 - PDF

Université de Jendouba Institut supérieur des langues appliquées et d'informatique de Béja BIG DATA 3LNSI A.U: 2024/2025  Chapitre 2 Principes fondamentaux de HadoopMapReduce ...

Université de Jendouba Institut supérieur des langues appliquées et d'informatique de Béja BIG DATA 3LNSI A.U: 2024/2025  Chapitre 2 Principes fondamentaux de HadoopMapReduce  Introduction Avant de parler de BigData, connaissez-vous les préfixes ? 4  Introduction Les mégadonnées ou Big Data sont des collections d’informations qui auraient été considérées comme gigantesques, impossible à stocker et à traiter, il y a une dizaine d’années. Internet : Google en 2015 : 10 Eo (10 milliards de Go), Facebook en 2014 : 300 Po de données (300 millions de Go), 4 Po de nouvelles données par jour, Amazon : 1 Eo. BigScience : télescopes (1 Po/jour), CERN (500 To/jour, 140 Po de stockage), génome, environnement... 5  Introduction Le traitement d’aussi grandes quantités de données impose des méthodes particulières. Un SGBD classique, même haut de gamme, est dans l’incapacité de traiter autant d’informations. Répartir les données sur plusieurs machines (jusqu’à plusieurs millions d’ordinateurs) dans des Data Centers système de fichiers spécial permettant de ne voir qu’un seul espace pouvant contenir des fichiers gigantesques et/ou très nombreux (HDFS), bases de données spécifiques (HBase, Cassandra, ElasticSearch). Traitements du type « map-reduce » : algorithmes faciles à écrire, exécutions faciles à paralléliser. → Tout ce module vise à enseigner l’outil Hadoop. 6  Hadoop : Définition Hadoop est une plateforme (framework) open source conçue pour réaliser d’une façon distribuée des traitements sur des volumes de données massives, de l’ordre de plusieurs pétaoctets. Ainsi, il est destiné à faciliter la création d’applications distribuées et échelonnables (scalables). Il s’inscrit donc typiquement sur le terrain du Big Data Hadoop est géré sous l’égide de la fondation Apache, il est écrit en Java Hadoop a été conçu par Doug Cutting en 2004 et été inspiré par les publications MapReduce, GoogleFS et BigTable de Google. 7  Hadoop : Définition Hadoop est un système de gestion de données et de traitements distribués. Il contient de beaucoup de composants, dont : HDFS un système de fichier qui répartit les données sur de nombreuses machines, YARN un mécanisme d’ordonnancement de programmes de type MapReduce. On va d’abord présenter HDFS puis YARN/MapReduce. 8 Section 1: Hadoop File System (HDFS) 9  Hadoop File System (HDFS) Présentation HDFS est un système de fichiers distribué. C’est à dire : les fichiers et dossiers sont organisés en arbre (comme Unix) ces fichiers sont stockés sur un grand nombre de machines de manière à rendre invisible la position exacte d’un fichier. L’accès est transparent, quelle que soient les machines qui contiennent les fichiers. les fichiers sont copiés en plusieurs exemplaires pour la fiabilité et permettre des accès simultanés multiples HDFS permet de voir tous les dossiers et fichiers de ces milliers de machines comme un seul arbre, contenant des Po de données, comme s’ils étaient sur le disque dur local. 10  Hadoop File System (HDFS) Organisation des fichiers Vu de l’utilisateur, HDFS ressemble à un système de fichiers Unix : il y a une racine, des répertoires et des fichiers. Les fichiers ont un propriétaire, un groupe et des droits d’accès. des répertoires pour les services Hadoop : /hbase, /tmp, /var un répertoire pour les fichiers personnels des utilisateurs : /user (attention, ce n’est ni /home, ni /users comme sur d’autres systèmes Unix). Dans ce répertoire, il y a aussi trois dossiers système : /user/hive, /user/history et /user/spark. un répertoire pour déposer des fichiers à partager avec tous les utilisateurs : /share Vous devrez distinguer les fichiers HDFS des fichiers « normaux ». 11  Hadoop File System (HDFS) Commande hdfs dfs La commande hdfs dfs et ses options permet de gérer les fichiers et dossiers : hdfs dfs -help hdfs dfs -ls [noms...] (pas d’option -l) hdfs dfs -cat nom hdfs dfs -mv ancien nouveau hdfs dfs -cp ancien nouveau hdfs dfs -mkdir dossier hdfs dfs -rm -f -r dossier (pas d’option -fr) Il faut toutefois noter que les commandes mettent un certain temps à réagir. D’autre part, nos machines ne sont pas très rapides. 12  Hadoop File System (HDFS) Échanges entre HDFS et le monde Pour placer un fichier dans HDFS, deux commandes équivalentes : hdfs dfs -copyFromLocal fichiersrc fichierdst hdfs dfs -put fichiersrc [fichierdst] Pour extraire un fichier de HDFS, deux commandes possibles : hdfs dfs -copyToLocal fichiersrc dst hdfs dfs -get fichiersrc [fichierdst] Exemple hdfs dfs -mkdir -p livres wget http://www.textfiles.com/etext/FICTION/dracula hdfs dfs -put dracula livres hdfs dfs -ls livres 13 hdfs dfs -get livres/center_ earth  Hadoop File System (HDFS) Comment fonctionne HDFS Comme avec de nombreux systèmes de fichiers, chaque fichier HDFS est découpé en blocs de taille fixe. Un bloc HDFS = 256Mo (à l’IUT, j’ai réduit à 64Mo). Selon la taille d’un fichier, il lui faudra un certain nombre de blocs. Sur HDFS, le dernier bloc d’un fichier fait la taille restante. Les blocs sont numérotés et chaque fichier sait quels blocs il occupe. Les blocs d’un même fichier ne sont pas forcément tous sur la même machine. Les répartir sur plusieurs machines permet d’y accéder simultanément par plusieurs processus. En plus, dans HDFS, il y a réplication des blocs sur plusieurs machines pour se prémunir contre les pannes. Chaque fichier se trouve donc en plusieurs exemplaires 14 à différents  Hadoop File System (HDFS) Organisation des machines pour HDFS Un cluster HDFS est constitué de machines jouant différents rôles exclusifs entre eux : L’une des machines est le maître HDFS, appelé le namenode. Cette machine contient tous les noms et blocs des fichiers, comme un gros annuaire téléphonique. Une autre machine est le secondary namenode, une sorte de namenode de secours, qui enregistre des sauvegardes de l’annuaire à intervalles réguliers. Certaines machines sont des clients. Ce sont des points d’accès au cluster pour s’y connecter et travailler. Toutes les autres machines sont des datanodes. Elles stockent les blocs du contenu des fichiers. 15  Hadoop File System (HDFS) Un schéma des nodes HDFS Les datanodes contiennent des blocs (notés A,B,C... ). Les mêmes blocs sont dupliqués (replication) sur différents datanodes, en général 3 fois. Cela assure : fiabilité des données en cas de panne d’un datanode, accès parallèle par différents processus aux mêmes données. Le namenode sait à la fois : sur quels blocs sont contenus les fichiers, sur quels datanodes se trouvent les blocs voulus. On appelle cela les metadata. Inconvénient majeur : panne du namenode = mort de HDFS, c’est pour éviter ça qu’il y a le secondary 16 namenode. Il archive les metadata, par exemple toutes les heures.  Hadoop File System (HDFS) Ecriture d’un fichier (1/2) Si on souhaite écrire un fichier au sein de HDFS , on va utiliser la commande principale de gestion de Hadoop, avec l’option fs. Mettons qu’on souhaite stocker le ficher page_livre.txt sur HDFS. Le programme va diviser le fichier en blocs de 64 Mo (ou autre, selon la configuration ) – supposons qu’on ait ici 2 Blocs. Il va ensuite annoncer au NameNode : Le NameNode va alors indiquer au programme qu’il doit stocker le bloc 1 sur le DataNode numéro 3, et le bloc 2 sur le DataNode numéro 1. Le client hadoop va alors contacter directement les DataNodes concernés et leur demander de stocker les deux blocs en question. Par ailleurs, les DataNodes s’occuperont – en informant le NameNode – de répliquer les données entre eux pour éviter toute perte de données. 17  Hadoop File System (HDFS) Ecriture d’un fichier (2/2) 18  Hadoop File System (HDFS) Lecture d’un fichier (1/2) Si on souhaite lire un fichier au sein de HDFS , on utilise là aussi le client Hadoop. Mettons qu’on souhaite lire le ficher page_livre.txt. Le client va contacter le NameNode, et lui indiquer < Je souhaite lire le fichier page_livre.txt>. Le NameNode lui répondra par exemple < il est composé de deux blocs. Le premier est disponible sur le DataNode 3 et 2, le second sur le DataNode 1 et 3>. Là aussi, le programme contactera les DataNodes directement et leur demandera de lui transmettre les blocs concernés. En cas d’erreur/non réponse d’un des DataNode, il passe au suivant dans la liste fournie par le NameNode. 19  Hadoop File System (HDFS) Lecture d’un fichier (2/2) 20  Hadoop File System (HDFS) Mode high availability Comme le namenode est absolument vital pour HDFS mais unique, Hadoop propose une configuration appelée high availability dans laquelle il y a 2 autres NameNodes en secours, capables de prendre le relais instantanément en cas de panne du namenode initial. Les NameNodes de secours se comportent comme des clones. Ils sont en état d’attente et mis à jour en permanence à l’aide de services appelés JournalNodes. Les NameNodes de secours font également le même travail que le secondary namenode, d’archiver régulièrement l’état des fichiers, donc ils rendent ce dernier inutile. 21  Hadoop File System (HDFS) API Java pour HDFS Hadoop propose une API Java complète pour accéder aux fichiers de HDFS. Elle repose sur deux classes principales : FileSystem représente l’arbre des fichiers (file system). Cette classe permet de copier des fichiers locaux vers HDFS (et inversement), renommer, créer et supprimer des fichiers et des dossiers FileStatus gère les informations d’un fichier ou dossier : taille avec getLen(), nature avec isDirectory() et isFile(), Ces deux classes ont besoin de connaître la configuration du cluster HDFS, à l’aide de la classe Configuration. D’autre part, les noms complets des fichiers sont représentés par la classe Path 22  Hadoop File System (HDFS) API Java pour HDFS : Exemple Voici quelques manipulations sur un fichier 23  Hadoop File System (HDFS) API Java pour HDFS : Lecture d’un fichier HDFS Voici un exemple simplifié de lecture d’un fichier texte 24  Hadoop File System (HDFS) API Java pour HDFS : Création d’un fichier HDFS voici comment créer un fichier 25  Hadoop File System (HDFS) API Java pour HDFS : Compilation et lancement Compiler et lancer ces programmes avec ce Makefile 26 YARN et MapReduce 27  YARN Qu’est-ce que YARN ? YARN (Yet Another Resource Negociator) est un mécanisme permettant de gérer des travaux (jobs) sur un cluster de machines. YARN permet aux utilisateurs de lancer des jobs MapReduce sur des données présentes dans HDFS, et de suivre (monitor) leur avancement, récupérer les messages (logs) affichés par les programmes. Éventuellement YARN peut déplacer un processus d’une machine à l’autre en cas de défaillance ou d’avancement jugé trop lent. En fait, YARN est transparent pour l’utilisateur. On lance l’exécution d’un programme MapReduce et YARN fait en sorte qu’il soit exécuté le plus rapidement possible 28  MapReduce Qu’est-ce que MapReduce ? MapReduce est un environnement Java pour écrire des programmes destinés à YARN. Java n’est pas le langage le plus simple pour cela, il y a des packages à importer, des chemins de classes à fournir... Il y a plusieurs points à connaître, c’est la suite de ce cours : Principes d’un job MapReduce dans Hadoop, Programmation de la fonction Map, Programmation de la fonction Reduce, Programmation d’un job MapReduce qui appelle les deux fonctions, Lancement du job et récupération des résultats. Commençons d’abord avec le type des données échangées entre Map et Reduce. 29  MapReduce Paires clé-valeurs Les données échangées entre Map et Reduce, et plus encore, dans la totalité du job sont des paires (clé, valeur) : une clé : c’est n’importe quel type de données : entier, texte... une valeur : c’est n’importe quel type de données Tout est représenté ainsi. Par exemple : un fichier texte est un ensemble de (n° de ligne, ligne). un fichier météo est un ensemble de (date et heure, température) C’est cette notion qui rend les programmes assez étranges au début : les deux fonctions Map et Reduce reçoivent des paires (clé, valeur) et émettent d’autres paires, selon les besoins de l’algorithme 30  MapReduce Map La fonction Map reçoit une paire en entrée et peut produire un nombre quelconque de paires en sortie : aucune, une ou plusieurs, à volonté. Les types des entrées et des sorties sont comme on veut. Cette spécification très peu contrainte permet de nombreuses choses. En général, les paires que reçoit Map sont constituées ainsi : la valeur de type text est l’une des lignes ou l’un des n-uplets du fichier à traiter la clé de type integer est la position de cette ligne dans le fichier (on l’appelle offset en bon français) Il faut comprendre que YARN lance une instance de Map pour chaque ligne de chaque fichier des données à traiter. Chaque instance traite la ligne qu’on lui a attribuée et produit des paires en sortie. 31  MapReduce Schéma de Map Tâches MAP et paires (clé, valeur) Les tâches MAP traitent chacune une paire et produisent 0..n paires. Il se peut que les mêmes clés et/ou valeurs soient produites. 32  MapReduce Reduce La fonction Reduce reçoit une liste de paires en entrée. Ce sont les paires produites par les instances de Map. Reduce peut produire un nombre quelconque de paires en sortie, mais la plupart du temps, c’est une seule. Par contre, le point crucial, c’est que les paires d’entrée traitées par une instance de Reduce ont toutes la même clé. YARN lance une instance de Reduce pour chaque clé différente que les instances de Map ont produit, et leur fournit uniquement les paires ayant la même clé. C’est ce qui permet d’agréger les valeurs. En général, Reduce doit faire un traitement sur les valeurs, comme additionner toutes les valeurs entre elles, ou déterminer la plus grande des valeurs... Quand on conçoit un traitement MapReduce, on doit réfléchir aux clés et valeurs nécessaires pour que ça marche. 33  MapReduce Schéma de Reduce Les tâches Reduce reçoivent une liste de paires ayant toutes la même clé et produisent une paire qui contient le résultat attendu. Cette paire en sortie peut avoir la même clé que celle de l’entrée. Tâches Reduce et paires (clé, valeur) 34  MapReduce Exemple Une entreprise de téléphonie veut calculer la durée totale des appels téléphoniques d’un abonné à partir d’un fichier CSV contenant tous les appels de tous les abonnés (n° d’abonné, n° appelé, date, durée d’appel). Ce problème se traite ainsi : 1. En entrée, on a le fichier des appels (1 appel par ligne) 2. YARN lance une instance de la fonction Map par appel 3. Chaque instance de Map reçoit une paire (offset, ligne) et produit une paire (n° abonné, durée) ou rien si c’est pas l’abonné qu’on veut. NB: l’offset ne sert à rien ici. 4. YARN envoie toutes les paires vers une seule instance de Reduce (car il n’y a qu’une seule clé différente) 5. L’instance de Reduce additionne toutes les valeurs des paires qu’elle reçoit et produit une seule paire en sortie (n° abonné, durée totale) 35  MapReduce Remarques En réalité, il n’y a pas qu’une seule instance de Reduce, il y en a plusieurs pour faire la réduction de manière hiérarchique plus rapidement. Car en général l’algorithme qu’on écrit dans la fonction Reduce est une boucle sur chaque valeur reçue. Également, en réalité, il n’y a pas une instance de Map par ligne de données. C’est la vision qu’on peut avoir en tant que programmeur, mais ça conduirait à un nombre gigantesque d’instances pour traiter un énorme fichier. En fait, YARN instancie un seul « Mappeur » par machine esclave et appelle sa méthode map à plusieurs reprises pour traiter les données séquentiellement. 36  MapReduce Étapes d’un job MapReduce Un job MapReduce comprend plusieurs phases : 1. Prétraitement des données d’entrée, ex: décompression des fichiers 2. Split: séparation des données en blocs traitables séparément et mise sous forme de (clé, valeur), ex: en lignes ou en n-uplets 3. Map: application de la fonction map sur toutes les paires (clé, valeur) formées à partir des données d’entrée, cela produit d’autres paires (clé, valeur) en sortie 4. Shuffle & Sort: redistribution des données afin que les paires produites par Map ayant les mêmes clés soient sur les mêmes machines 5. Reduce: agrégation des paires ayant la même clé pour obtenir le résultat final. 37  MapReduce Un schéma 38  MapReduce Explication du schéma 1. Au début, YARN se renseigne sur l’emplacement des données auprès du NameNode et les fait décompresser si besoin par les datanodes concernés. 2. La phase Split consiste à construire des paires (n° de n-uplet, n-uplet) à fournir aux tâches Map. 3. YARN crée des processus Map sur chaque machine contenant une partie des données et leur fournit les paires de leur machine successivement. 4. Chaque tâche Map analyse ses données et émet ou non une paire. Ça peut consister à convertir des chaînes en nombres, à faire des calculs, etc. 5. YARN trie les paires sortant de Map selon leur clé et les envoie sur la machine qui fait tourner la tâche Reduce concernée par cette clé. 6. Les tâches Reduce reçoivent une liste de paires et effectuent la réduction des valeurs (max, sum, avg... ). Elles émettent seulement la valeur finale. Elles 39 peuvent être mises en cascade quand il y a beaucoup de paires. Mise en œuvre dans Hadoop 40  Mise en œuvre dans Hadoop Présentation ▪ On arrive à la partie la plus technique : la programmation d’un job MapReduce en Java. Il faut définir trois classes : 1. Une sous-classe de Mapper. Elle contient une seule méthode, appelée map qui reçoit une paire clé-valeur en paramètre. Elle génère un nombre quelconque de paires. 2. Une sous-classe de Reducer. Elle contient également une seule méthode, appelée reduce qui reçoit une liste de paires en paramètre. Elle génère une seule paire. 3. Une classe générale qui crée un Job faisant référence aux deux précédentes classes. ▪ Les deux premières sont des patrons (templates) paramétrées par les types des clés et des valeurs. 41  Mise en œuvre dans Hadoop Squelette de Mapper 42  Mise en œuvre dans Hadoop Explications ▪ La classe Mapper est paramétrée par 4 types. Hélas, ce ne sont pas les types standard de Java, mais des types spéciaux permettant de transmettre efficacement des données entre les différents ordinateurs du cluster. Ça complique légèrement les programmes. 43  Mise en œuvre dans Hadoop Types de données MapReduce ▪ Les types Text, IntWritable... sont des implémentations d’une interface appelée Writable. Cette interface comprend : ▪ un constructeur. On peut mettre la valeur initiale en paramètre. IntWritable val = new IntWritable(34); ▪ un modificateur : void set(nouvelle valeur); val.set(35); ▪ un accesseur : type get() int v = val.get(); 44  Mise en œuvre dans Hadoop Interface Writable ▪ Elle permet la sérialisation, c’est à dire l’écriture d’une structure de données sous forme d’octets et l’opération inverse, la désérialisation qui permet de reconstruire une structure de données à partir d’octets. ▪ La sérialisation est nécessaire pour échanger des données entre machines. Cela fait partie de la technique appelée Remote Procedure Call (RPC). On ne peut pas simplement échanger les octets internes car les machines du cluster ne sont pas obligatoirement toutes pareilles : nombre d’octets, ordre des octets... ▪ Cette interface n’est pas limitée à des types simples mais peut gérer des collections (tableaux, listes, dictionnaires... ) et classes. 45  Mise en œuvre dans Hadoop Classe Text ▪ La classe Text permet de représenter n’importe quelle chaîne. Elle possède quelques méthodes à connaître : 1. String toString() extrait la chaîne Java 2. int getLength() retourne la longueur de la chaîne 3. int charAt(int position) retourne le code UTF8 (appelé point) du caractère présent à cette position ▪ Ces méthodes ne sont pas suffisantes. Il faudra souvent convertir les Text en chaînes. 46  Mise en œuvre dans Hadoop Squelette de Reducer 47  Mise en œuvre dans Hadoop Explications ▪ La méthode reduce reçoit une collection de valeurs venant du Mapper. CleI et ValeursI sont les clés et valeurs intermédiaires. Il faut itérer sur chacune pour produire la valeur de sortie du réducteur. ▪ Comme pour map, la classe est paramétrée par les types des clés et des valeurs à manipuler. Ce sont des Writable : Text, IntWritable... ▪ Une chose cruciale n’est pas du tout vérifiée par Java : il est obligatoire que les types des clés TypCleI et valeurs d’entrée TypValI du réducteur soient exactement les mêmes que les types des clés et valeurs de sortie du mapper. Si vous mettez des types différents, ça passera à la compilation mais plantera à l’exécution. 48  Mise en œuvre dans Hadoop Squelette de Traitement ▪ Voici la classe principale qui crée et lance le job MapReduce 49  Mise en œuvre dans Hadoop Squelette de Traitement (cœur) ▪ La méthode run contient ceci : 50  Mise en œuvre dans Hadoop Explications ▪ La méthode run est chargée de créer et lancer un Job. Il faut noter que la spécification Hadoop a beaucoup changé depuis les premières versions. Il faut actuellement faire ainsi : 1. Obtenir une instance de Configuration. Elle contient les options telles que les formats des fichiers, leur nom HDFS complet, leur codec de compression. 2. Créer un Job, lui indiquer les classes concernées : mapper et reducer. 3. Fournir les noms complets des fichiers à traiter et à produire. 4. Indiquer les types des clés et valeurs. Par défaut, ce sont des Text. 5. Attendre la fin du job et retourner un code d’erreur. 51  Mise en œuvre dans Hadoop Compilation et lancement d’un traitement 1. Compilation → hadoop com.sun.tools.javac.MainTraitement*.java 2. Emballage dans un fichier jar. NB: c’est plus compliqué quand il y a des packages. → jar cfeTraitement.jarTraitementTraitement*.class 3. Préparation : mettre en place les fichiers à traiter, supprimer le dossier de sortie → hdfs dfs -rm -r -f sortie 4. Lancement → yarn jarTraitement.jar entree sortie 5. Résultats dans le dossier sortie → hdfs dfs -cat sortie/part-r-00000 52