Utilisateur:Lpalmie/MapReduce et son implémentation

• Gérer mon brouillon
• Demander une relecture (voir les demandes)
• Publier mon brouillon

Cette page est un brouillon appartenant à Lpalmie

Conseils de rédaction

• → N'hésitez pas à publier sur le brouillon un texte inachevé et à le modifier autant que vous le souhaitez.
• → Pour enregistrer vos modifications au brouillon, il est nécessaire de cliquer sur le bouton bleu : « Publier les modifications ». Il n'y a pas d'enregistrement automatique.

Si votre but est de publier un nouvel article, votre brouillon doit respecter les points suivants :

• Respectez le droit d'auteur en créant un texte spécialement pour Wikipédia en français (pas de copier-coller venu d'ailleurs).
• Indiquez les éléments démontrant la notoriété du sujet (aide).
• Liez chaque fait présenté à une source de qualité (quelles sources  – comment les insérer).
• Utilisez un ton neutre, qui ne soit ni orienté ni publicitaire (aide).
• Veillez également à structurer votre article, de manière à ce qu'il soit conforme aux autres pages de l'encyclopédie (structurer – mettre en page).

• → Si ces points sont respectés, pour transformer votre brouillon en article, utilisez le bouton « publier le brouillon » en haut à droite. Votre brouillon sera alors transféré dans l'espace encyclopédique.

Le MapReduce est un modèle de programmation. Il permet de manipuler de grande quantité de données en les distribuant à traiter dans un cluster de machines. Ce modèle est en train de connaitre un vif succès auprès des sociétés possédant d'importants datacenters telles que Amazon ou Facebook, il commence aussi à être utilisé au sein du Cloud Computing. De nombreux frameworks ont vu le jour afin d'implémenter le MapReduce. Le plus connue est Hadoop qui a été programmé par Yahoo!. Ce framework possède des inconvénients qui réduisent considérablement ses performances notamment en milieu hétérogène. Des frameworks qui permettent d'améliorer les performances de Hadoop ou les performances globales du MapReduce (tant en terme de vitesse de traitement qu'en consommation électrique) commencent à voir le jour.

Présentation

Un modèle de programmation

Le MapReduce est un modèle de programmation popularisé par Google. Il est principalement utilisé pour la manipulation et le traitement d’un nombre important de données au sein d’un cluster de nœuds. Le MapReduce consiste en deux fonctions map() et reduce().

• La fonction map prend en entrée une paire <clé,valeur> afin de générer un ensemble de paires <clé,valeur> intermédiaires.

map(clé1,valeur1) --> list(clé2,valeur2)

//En pseudo code cela donnerait
Map(void * document){
   entier cles = 1;
   for each mot m in document
      calculIntermediaire(m,cles);
}

• La fonction reduce fusionne toutes les valeurs intermédiaires associées avec la même clé intermédiaire

reduce(key2,list(valeur2))--> list(valeur2)

//En pseudo code cela donnerait
Reduce(entier cles, Iterator values){
 int result = 0;
 for each v in values
    result += v;
}

Un cluster MapReduce utilise une architecture de type Maître-esclave où un nœud maître dirige tous les nœuds esclaves.

MapReduce possède quelques caractéristiques  :

• Le modèle de programmation du MapReduce est simple mais très expressif. Bien qu’il ne possède que deux fonctions, map() et reduce(), elles peuvent être utilisées pour de nombreux types de traitement des données, les fouilles de données, les graphes … Il est indépendant du système de stockage et peut manipuler de nombreux types de variable.
• Le système découpe automatiquement les données en entrée en bloc de données de même taille. Puis, il planifie l’exécution des tâches sur les nœuds disponibles.
• Il fournit une tolérance aux fautes à grain fin grâce à laquelle il peut redémarrer les nœuds ayant rencontrés une erreur ou assigner la tâche à un autre noeud.
• La parallélisation est invisible à l'utilisateur afin de lui permettre de se concentrer sur le traitement des données

Schéma de fonctionnement du MapReduce

Une fois qu'un noeud a terminé une tâche, on lui assigne un nouveau bloc de données. Grâce à cela, un noeud rapide fera beaucoup plus de calculs qu'un noeud plus lent. Le nombre de tâches Map ne dépend pas du nombre de noeuds, mais du nombre de blocs de données en entrée. Chaque bloc se fait assigner une seule tâche Map. De plus, toutes les tâches Map n'ont pas besoin d'être exécutées en même temps en parallèle, les tâches Reduce suivent la même logique. Par exemple, si des données en entrée sont divisées en 400 blocs et qu'il y a 40 noeuds dans le cluster, le nombre de tâches Map sera de 400. Il faudra alors 10 vagues de Map pour réaliser le mapping des données .

Le MapReduce est apparu en 2004, c'est une technologie encore jeune qui possède quelques points faibles  :

• Il ne supporte pas les langages haut niveau comme le SQL
• Il ne gère pas les index. Une tâche MapReduce peut travailler après que les données en entrée soient stockées dans sa mémoire. Cependant, MapReduce a besoin de parser chaque donnée en entrée afin de la transformer en objet pour la traiter, ce qui provoque des baisses de performance.
• Il utilise un seul flot de données. Le MapReduce est facile à utiliser avec une seule abstraction mais avec un flot de donnée fixe. Par conséquence, certains algorithmes complexes sont difficiles à implémenter avec seulement les méthodes map() et reduce(). De plus, les algorithmes qui requièrent de multiples éléments en entrée ne sont pas bien supportés car le flot de données du MapReduce est prévu pour lire un seul élément en entrée et génère une seule donnée en sortie.
• Quelques points peuvent réduire les performances du MapReduce. Avec sa tolérance aux pannes et sa scalabilité, les opérations du MapReduce ne sont pas toujours optimisées pour les entrées/sorties. De plus, les méthodes map() et reduce() sont bloquantes. Cela signifie que pour passer à l'étape suivante, il faut attendre que toutes les tâches de l'étape courante soient terminées. Le MapReduce n'a pas de plan spécifique d'exécution et n'a pas optimisé le transfert de données entre ces noeuds.

Les facteurs de performance

D'après une étude , il existe cinq facteurs qui influent sur les performances du MapReduce.

Le modèle de programmation MapReduce a été conçu pour qu'il soit indépendant du système de stockage des données. Il lit les paires <clé,valeur> à l'aide d'un lecteur. Le lecteur récupère chaque enregistrement contenu dans le système de stockage, puis il les place dans une paire <clé,valeur> afin de pouvoir les traiter lors du MapReduce. Bien que le MapReduce ne dépende pas d'un système de stockage particulier, il éprouve des difficultés lorsque le système de stockage est une base de données. Les systèmes de bases de données parallèles commerciaux utilisent un moteur pour exécuter les requêtes et moteur de stockage. Pour exécuter une requête, le moteur de requêtes lit directement les données depuis le moteur de stockage. Le MapReduce doit d'abord lire la valeur, puis la stocker dans une paire à l'aide du lecteur, ce qui explique pourquoi le MapReduce est moins performant dans les bases de données. En comparant le MapReduce et les systèmes de bases de données parallèles, trois facteurs, pouvant affecter les performances du MapReduce, ont été mis en évidence :

• Mode entrée/sortie : la façon dont le lecteur récupère les informations
• Le parsage des données : la façon dont le lecteur parse les données
• L'indexation

Le MapReduce utilise un système de planification afin d'assigner les blocs de données aux nœuds disponibles dans le cluster. Ce système provoque des coûts d'exécution et peut ralentir l'exécution du MapReduce. Deux facteurs peuvent affecter Les performances du MapReduce :

• La taille des blocs de données distribués
• L'algorithme de planification

Mode entrée/sortie

Deux modes de lectures peuvent être utilisés par le lecteur afin de lire les données stockées dans un système de stockage. Le mode entrée/sortie direct avec lequel le lecteur lit directement les données stockées dans le disque local. Dans ce cas, les données sont transférées depuis la mémoire cache vers la mémoire du lecteur en utilisant l'accès direct à la mémoire. Le mode entrée/sortie streaming avec lequel le lecteur lit les données depuis un autre processus en cours d'exécution à l'aide de moyen de communication entre les processus comme TCP/IP ou JDBC.

D'après les tests réalisés dans cette étude , les différences de performance entre ces deux modes est faible (environ 10%).

Le parsage des données

Quand le lecteur récupère les données depuis le système de stockage, il doit convertir les données en paire <clé,valeur> pour continuer l'exécution (ce processus s'appelle le parsage de données). Le parsage consiste à décoder les données depuis leur format natif de stockage afin de les transformer vers un format qui pourra être utilisé par un langage de programmation. Il existe deux types de décodage, le décodage immuable et le décodage mutable. Le décodage immuable consiste à transformer des données en objet immuable. Les objets immuables en programmation orientée objet est un objet dont l'état ne peut être modifié après sa création contrairement aux objets mutables.

Lorsque l'on utilise le décodage immuable, chaque donnée sera alors placé dans un objet immuable. Par conséquent, si l'on décode 4 millions de données, 4 millions d'objets immuables seront créés. Par défaut, le MapReduce de Google utilise des objets immuables.

Une autre méthode consiste à utiliser le décodage mutable. Avec ce décodage, un objet mutable est réutilisé pour décoder toutes les données. Ainsi, le nombre de données n'est plus important car seulement un objet sera créé.

D'après les études , les faibles performances du parsage sont causées à cause du décodage immuable. Le décodage immuable est plus lent que le décodage mutable car il produit un grand nombre d'objets immuables durant le processus de décodage. La création de tous ces objets immuables augmentent la charge de travail des processeurs.

L'indexation

Le MapReduce étant indépendant du système de stockage, il ne peut pas prendre en compte l'ensemble des données en entrée pour avoir un index disponible. MapReduce ne semble pas capable d'utiliser des index. Cependant, trois méthodes pour utiliser les index ont été trouvées afin d'augmenter la vitesse du processus du traitement des données.

• MapReduce offre une interface aux utilisateurs pour spécifier un algorithme pour la distribution des nœuds. Par conséquent, il est possible d'implémenter un algorithme qui utilise l'index pour réduire les blocs de données.

• Si l'ensemble des données en entrée du MapReduce est un ensemble de fichiers indexés (avec des B-Arbres), on peut traiter ces données en implémentant un nouveau lecteur. Le lecteur va prendre certaines conditions de recherche et va l'appliquer à l'index afin de récupérer les données contenues dans chaque fichier.

• Si les données en entrée du MapReduce consiste en des tables indexées stockées dans n serveurs de bases de données, il est possible d'appliquer n tâches map pour traiter ces tables. Dans chaque tâche map, la fonction map() envoie une requête SQL vers un serveur de base de données pour récupérer les données. Ainsi, on utilise de façon transparente les index de la base de données.

Distribution des blocs/Algorithme de planification

Pour réduire les temps de planification, il est possible de modifier la taille des blocs de données à distribuer aux noeuds du cluster. Si l'on augmente la taille des blocs de données, alors la planification sera plus rapide, car elle nécessitera moins de tâches map. Cependant, si l'on augmente trop la taille des blocs, alors le risque d'échec est plus important.

Hadoop

Présentation

Hadoop est une implémentation open source en Java du MapReduce distribué par la fondation Apache. Il a été mis en avant par des grands acteurs du web tels queYahoo! et Facebook . Les deux caractéristiques principales de Hadoop sont le framework MapReduce et le Hadoop Distributed File System (qui s’inspire du GFS ou Google File System). Le HDFS permet de distribuer les données et de faire des traitements performants sur ces données grâce au MapReduce en distribuant une opération sur plusieurs nœuds afin de la paralléliser .

Dans Hadoop, le nœud maître est appelé le JobTracker et les nœuds esclaves sont appelés TaskTracker. Chaque nœud esclave va contenir les blocs de données en les répliquant. Le nœud maître connaît les emplacements des différentes répliques. Le nœud maître secondaire sert à effectuer des sauvegardes régulières du nœud maître afin de pouvoir le réutiliser en cas de problème.

Hadoop exécute une tâche de type MapReduce en commençant par diviser les données en entrée en bloc de données de même taille. Ensuite, chaque bloc est planifié pour être exécuté par un TaskTracker. Le processus d’assignement des tâches est implémenté comme un protocole de type « battement de cœur ». Cela signifie que le TaskTracker notifie le JobTracker que sa tâche est terminée afin que celui-ci lui assigne une nouvelle tâche à exécuter. Lorsque la fonction map est achevée, le système va regrouper toutes les paires intermédiaires et lancer une série de réductions pour produire le résultat final.

Schéma du Hadoop Distributed File System

Performances

En milieu hétérogène

Aujourd'hui, l'implémentation de Hadoop considère que le traitement se réalise sur un cluster de machines homogènes (c'est-à-dire qu'elles possèdent toutes les mêmes caractéristiques matérielles). Il ne tient pas compte non plus de la localité des données, il considère que toutes les données sont locales. Malheureusement, ces 2 facteurs peuvent influencer les performances du MapReduce de manière conséquente.

En milieu homogène, tous les noeuds ont la même charge de travail, ce qui indique qu'aucune donnée ne devra être transférée d'un noeud vers un autre. En milieu hétérogène, un noeud ayant des performances élevées peut terminer son traitement local plus rapidement qu'un noeud ayant des performances plus faibles. Lorsque que le noeud rapide a terminé son traitement, il devra récupérer les données non traitées d'un ou plusieurs autres noeuds plus lents. Le transfert d'une donnée d'un noeud lent vers un noeud rapide a un coût élevé.

Dans le Cloud

Présentation

Le MapReduce a émergé en 2004 comme un important modèle de programmation pour les applications utilisant d’énormes quantités de données grâce à sa répartition efficace du travail sur différents noeuds de calcul. Il commence notamment à être utilisé dans le Cloud Computing car son nombre de données stockées et manipulées ne cesse de croître. Il est donc nécessaire d'avoir un moyen d'améliorer le traitement des données au sein du Cloud.

En termes de ressources, le Cloud peut se caractériser en 3 points : une capacité de stockage des données importante, de la puissance de calcul à la demande, utilise peu de bande passante. La taille des données stockées dans le Cloud augmente sans cesse notamment les fichiers de type image, vidéo, audio ou les instruments scientifiques pour réaliser des simulations . Le traitement de ces données est devenu l'un des principaux défis du Cloud Computing.

Le Cloud utilise principalement des machines virtuelles(VM) pour exécuter les applications. Les VM permettent d’exploiter pleinement les ressources du système, d’améliorer la fiabilité du système en sauvegardant l'état du système à l'aide de fonctionnalité incluse dans les machines virtuelles et de réduire la consommation énergétique en réduisant le nombre de noeuds utilisés pour réaliser les tâches.

Combiner le Cloud Computing qui utilise des VM et le MapReduce peut s'avérer être intéressant. Principalement, parce que les technologies de virtualisation sont arrivées à maturité . Elles ont déjà été utilisées pour des grilles de calcul pour utiliser pleinement les ressources des noeuds de calcul, des applications HPC (High-Performance computing) . De plus, d’un côté il y a la progression croissante de la popularité du Cloud (qui utilise des VM), de l’autre, le MapReduce commence à être largement utilisé grâce à ses nombreux points forts notamment pour sa scalabilité et sa tolérance aux fautes. C’est pourquoi, combiner ces deux technologies permettrait de créer un moyen efficace de traiter des données conséquentes sur le Cloud. A cela s’ajoute le fait que le MapReduce utilise des tâches spéculatives. Ce sont des tâches qui peuvent être relancées sur un autre noeud si elles sont détectées comme étant trop lente. Le problème est que relancer une tâche peut faire diminuer les performances en augmentant le temps d'exécution étant donné que la tâche doit reprendre tout le traitement. Grâce aux nouvelles technologies des VM que sont la sauvegarde et la migration, les performances et la fiabilité du MapReduce sont améliorées. La fonctionnalité de sauvegarde consiste à sauvegarder l’état du système, il peut ainsi retourner à cet état s'il rencontre une erreur grave qui empêche son bon fonctionnement. La fonctionnalité de migration consiste à distribuer une VM ou une application sur plusieurs nœuds physiques sans stopper l’application.

Performance

Pour tester le MapReduce, les différentes mesures sont effectuées sur le framework Hadoop. Des tests de performance sont réalisés sur le HDFS car il joue un grand rôle dans l’exécution du MapReduce. En effet, c'est notamment le HDFS qui se charge de la répartition des tâches sur les différents noeuds, il décide aussi de la taille des données à traiter par noeud. D’autres tests seront effectués sur la faisabilité d’utiliser des VM pour améliorer les performances du MapReduce en augmentant l’utilisation des ressources (en augmentant le nombre de VM par noeud). Les performances sont mesurées à l’aide de deux benchmark connus, le benchmark de tri et de comptage de mots.

benchmark de tri

ce benchmark utilise le map/reduce framework pour trier des données en entrée.

Le benchmark « Comptage de mots »

ce benchmark compte le nombre d’occurrences de chaque mot dans un fichier et écrit le résultat sur le disque local.

HDFS

L’évaluation des performances du HDFS se fait dans un cluster physique (PH-HDFS) et dans un cluster virtuel (VM-HDFS lors d’écriture et de lecture de données afin de montrer les différences de performances entre un cluster physique et un cluster virtuel.

Temps d'écriture et de lecture en fonction de la taille du fichiers - Cluster Physique vs Cluster virtuel

Dans cette évaluation, plusieurs transferts de données de différentes tailles sont effectués. Le PH-HDFS réalise des meilleurs temps, les écarts se creusent lorsque les données augmentent en taille.

Temps d'écriture et de lecture en fonction du nombre de requêtes - Cluster Physique vs Cluster virtuel

Dans cette évaluation, une ou plusieurs requêtes sont lancées simultanément et on mesure le temps nécessaire pour réaliser le transfert de données. Les performances du PH-HDFS sont meilleurs que le VM-HDFS.

VMs feasiblity

Avec l’avènement des processeurs multi-cœurs, il est pertinent de tester d’utiliser des processeurs multi-cœurs du cluster pour installer plusieurs VM par processeur afin d'utiliser pleinement les capacités du noeud. Pour les tests, plusieurs clusters ont été mise en place afin d'évaluer l'impact de l'augmentation du nombre de VM par noeud :

• Ph-Cluster : 7 nœuds physiques
• V-Cluster : 1 VM par nœud – 7 nœuds VM dans le cluster
• V2-Cluster : 2 VM par nœud – 13 nœuds VM dans le cluster
• V4-Cluster : 4 VM par nœud – 25 nœuds VM dans le cluster

Temps de traitement du WordCount en fonction du cluster et de la taille des fichiers - Cluster Physique vs Cluster virtuel

Comme le montre cette figure, le benchmark « comptage de mots » avec le Ph-Cluster est plus rapide que le V-Cluster. Lorsque les données font 1 Gb, la différence est faible. Mais lorsque les données sont beaucoup plus importantes (ici 8 Gb), on observe un écart important. Cela est causé par l’augmentation des tâches spéculatives ce qui cause une utilisation inefficace des ressources. En revanche, les clusters V2 et V4 sont beaucoup plus performants que le Ph-Cluster car il y a beaucoup plus de calculs par cycle.

Temps de traitement du WordCount et du Sort en fonction du cluster et de la taille des fichiers - Cluster Physique vs Cluster virtuel

Lorsque le benchmark de tri est lancé, le temps d’exécution, pour la même distribution des données, augmente avec l’augmentation du nombre de VM déployées sur un nœud physique. De plus, les écarts se creusent lorsque la taille des données distribuées augmente. Cela est provoqué pour 3 raisons : les mauvaises performances du HDFS dans les VM lors des opérations de lecture/écriture (comme vu précédemment), l’augmentation du nombre de tâche spéculative (Figure 6) et l’importante quantité de données transférées pendant les étapes intermédiaires.

Nombre de tâches spéculatives en fonction des données sur WordCount et Sort - Cluster Physique vs Cluster virtuel

Extensions de Hadoop

Rappelons que Hadoop est un framework basé sur le modèle de programmation MapReduce. Etant très utilisé dans les calculs de très grandes masses de données, plusieurs améliorations de ce framework sont apparues. Ces extensions sont des frameworks : "BlobSeer" modifie le système de fichiers pour améliorer l’accès aux données. "Phoenix" répartit les tâches sur les processeurs en utilisant le MapReduce, et "Mars" améliore le calcul de données sur des processeurs graphiques.

Amélioration de Hadoop en milieu hétérogène

Présentation

Comme dit précédemment, le problème de performance de Hadoop en milieu hétérogène est dû au transfert des données entre les noeuds rapides et les noeuds lents. Lorsqu'une large quantité de données est transférée, cela affecte de manière significative.

Afin d'améliorer les performances de Hadoop, il faut minimiser le transfert des données entre les noeuds rapides et lents. Pour cela, un mécanisme de placement de données a été implémenté; il distribue et stocke les données à travers de nombreux noeuds hétérogènes en fonction des capacités de leurs capacités. Avec cela, le transfert des données peut être réduit si le nombre de données placées dans le disque de chaque noeud est proportionnel à la vitesse de traitement des noeuds.

La réplication des données présente de nombreuses limitations. Premièrement, cela représente un coût important pour créer chaque réplique des données à l'intérieur d'un cluster ayant un nombre important de noeuds. Deuxièmement, distribuer un nombre important de répliques peut provoquer une surcharge de la bande passante du cluster. Troisièmement, stocker les répliques requiert des disques ayant une énorme quantité de stockage.

Amélioration

Des chercheurs se sont focalisés sur ce problème afin de produire un mécanisme de placement des données. Pour corriger le problème, ils se sont penchés sur le meilleur moyen de placer les données où les fichiers ne seront découpés et distribués sur plusieurs noeuds sans les dupliquer.

Ce mécanisme est basé sur 2 algorithmes qui sont incorporés dans le HDFS de Hadoop :

• Le premier algorithme consiste à distribuer distribuer des fragments du fichier en entrée.
• Le deuxième algorithme consiste à réorganiser les fragments du fichier et à corriger les erreurs qui peuvent arriver après l'exécution du premier algorithme.

Le premier algorithme fonctionne comme ceci :

• il commence par diviser le fichier d'entrée en un nombre de fragments de même taille.
• Puis il assigne les fragments aux noeuds du cluster en fonction de la vitesse de traitement des noeuds. (Cela aura pour conséquence qu'un noeud avec des faibles performances aura moins de fragments à traiter qu'un noeud avec des meilleures performances).

Si l'on considère une application, utilisant le MapReduce, et un fichier d'entrée dans un cluster hétérogène de Hadoop. Le placement initial des données ne tient pas compte des performances des noeuds car Hadoop estime que tous les noeuds vont exécuter et terminer leur tâche avec, environ, le même temps. Des expérimentations , on montrait que en règle général, le temps de traitement de chaque noeud était stable car le temps de réponse de chaque noeud est linéairement proportionnel à la taille des données. Avec ceci, on peut quantifier la vitesse de traitement de chaque noeud dans un environnement hétérogène. Le terme pour définir la performance de chaque noeud est "Ratio Performance".

Le Ratio Performance de chaque noeud est déterminé à l'aide de cette procédure :

• Les opérations d'une application utilisant le MapReduce sont réalisées séparément sur chaque noeud
• On récupère ensuite les temps de réponse de chaque noeud
• Le temps de réponse le plus court est utilisé comme temps pour normaliser les mesures du temps de réponse
• Les valeurs normalisées, appelées Ratio Performance, sont utilisées par l'algorithme de placement pour distribuer les fragments de fichier aux noeuds.

Prenons un exemple afin de montrer comment les Ratio Performance sont calculées. Prenons 3 noeuds hétérogènes A, B et C dans un cluster Hadoop. Après avoir exécuté l'application séparément sur chaque noeud, on récupère le temps de réponse de chaque noeud A, B et C (10 sec, 20 sec et 30 sec respectivement). Le temps de réponse du noeud A est le plus court, par conséquence, le Ratio Performance de A vaut 1. Les Ratio Performance de B et C valent 2 et 3 respectivement. Cela signifie que le noeud A pourra gérer 30 fragments du fichier d'entrée tandis que le noeud C en gérera seulement 10.

Après l'exécution de l'algorithme de distribution des fragments pour réaliser le placement initial, les fragments peuvent être corrompus pour plusieurs raisons:

• de nouvelles données ont été ajoutées au fichier de départ,
• des blocs de données ont été supprimés du fichier,
• de nouveaux noeuds ont été ajoutés au cluster.

Pour éviter ces problèmes, un deuxième algorithme a été implémenté pour réorganiser les fragments du fichier en le basant sur le Ratio Performance des noeuds.

Cet algorithme fonctionne comme ceci :

• Les informations concernant la topologie du réseau et l'espace disque du cluster sont collectées par le serveur de distribution des données.
• Le serveur créé deux listes de noeuds : une liste de noeuds qui contient le nombre de fragments dans chaque noeud qui peuvent être ajoutés , une liste de noeuds qui contient le nombre de fragments locals dans chaque noeud qui excédent la capacité de celui-ci.
• Le serveur de distribution des données transfère les fragments d'un noeud, dont la capacité du disque a été dépassé, vers un noeud qui possède encore de l'espace disque.

Dans le processus de migration des données entre 2 noeuds (un surchargé et un ayant de l'espace disque), le serveur transfère les fragments du fichier depuis le noeud source de la liste des noeuds en surcharge vers un noeud ayant encore de l'espace de la liste des noeuds sous-utilisés. Ce processus est répété jusqu'à ce que le nombre de fragments dans chaque noeud soit en accord avec son Ratio Performance.

Performances

Afin de montrer les résultats de ces algorithmes de cette étude, deux tests ont été effectués, le Grep et le WordCount. Ces 2 applications tournent sur des clusters Hadoop. Grep est un outil de recherche pour une expression régulière dans un texte. WordCount est un programme utilisé pour compter les mots dans un texte.

Les résultats ont montré que ces algorithmes ont amélioré les performances du Grep de 17% en moyenne et le WordCount de 7% en moyenne.

BlobSeer

Présentation

Hadoop est un framework basé sur le modèle de programmation Map/Reduce. Et pour cela Hadoop doit s'appuyer sur un "puissant outil" : son système de fichier HDFS . Le système de fichier HDFS(appelé Hadoop File Système) est un paramètre important de la performance de calcul de Hadoop car il lui est spécifique. C'est à dire qu'il a été conçut de façon à maximiser les accès aux données.

Seulement des erreurs subsistent lorsque plusieurs processus accèdent à un même fichier. C'est le framework Blobseer qui propose une solution à ce problème. En effet, ce framework propose de modifier le HDFS pour le remplacer par son propre système de fichier le BSFS (Blobseer File System).

Un nouveau système de fichier

Les raisons du changement de ce système de fichier sont liés aux problèmes d’accès concurrent sur un même fichier. Le système HDFS a été conçut de manière à permettre les meilleurs performance de calcul de Hadoop, cependant cette implémentation ne suffit pas. Le système de fichier HDFS ne permet pas de maintenir un haut débit sur les accès concurrent à un même fichier, de plus le système de fichier actuel ne permet pas certaines fonctionnalités tel que le "versioning" ou encore différents update simultanés sur un même fichier.

L'un des points fort de Hadoop est de parcourir des Pétaoctets en quelques heures seulement, cela est du au fait que les fichiers ont une taille de plusieurs centaines de Gigaoctets. De ce fait, il devient possible d'accéder à de petites parties d'un même fichier de façons concurrente. Il ne faut pas oublier qu'il serait impossible de travailler avec des millions de petits fichiers au lieu d'un seul, et même si le système de fichier le permet, maintenir un très haut débit n'est pas faisable.

Blobseer propose donc un système de fichier qui permet d'accéder à des petites parties d'un grand fichier, permettant ainsi à des milliers de "client" de modifier le même fichier sans problèmes de conflit. BSFS permet donc aussi le "versioning" ce qui permet de supprimer les changements indésirable et la création de branches indépendantes ce qui ne devrait pas diminuer les performances de calcul ni la surcharge de l'espace de stockages qui serait du à des milliers de petits fichiers.

Performances

Les performances du framework ont été testés sur grid5000 . Blobseer a été comparé avec Hadoop. Pour cela, les frameworks ont été utilisé sur des micro-tâches :

• Un seul processus qui écrit sur un fichier volumineux( > 20 GB).
• Des processus qui lisent les mêmes parties d'un seul fichier en même temps.
• Des processus qui écrivent dans un même fichier volumineux.

Les relevés de performances ont montré que le débit (l’accès au données) était jusqu'a 2 fois supérieur à celui de Hadoop. Ces résultats ont aussi révélé que le framework BlobSeer peut supporter jusqu’à 2 fois plus de client (c'est-à-dire le nombre d’accès sur un seul fichier). Le ratio des performances, entre BlobSeer et Hadoop n'est pas supérieur à 2. En effet, BlobSeer utilise les mêmes capacités de calcul que Hadoop, à la différence que le système de fichier a été modifié.

Phoenix

Présentation

Phoenix est une API basée sur le modèle MapReduce, proposé par google. La différence est que phoenix est utilisé sur les ordinateurs multi-coeurs et donc, il n'utilise pas des serveurs mais des threads pour pouvoir utiliser le MapReduce. Phoenix est basé sur un langage fonctionnel, de façons à rendre la parallélisation totalement transparente à l'utilisateur. Phoenix est utilisé en c et c++.

La structure

C'est une api développée pour être utilisée avec du code c/c++ mais elle peut être facilement exportable en java/c# . Pointeurs, Buffer, P-Threads sont utilisés pour augmenter la performance de l'api. Les Buffers permettent de d'accéder aux données dans une mémoire partagée. Il y a deux types de buffers, les input/output sont les buffers contenant les données en entrée ainsi que celles en sortie, c'est à dire les données dont l'utilisateur a besoin, et les autres buffers. Ces derniers sont ceux utilisés pour réaliser le Map/reduce, ce sont donc des buffers "invisibles" aux yeux de l'utilisateur. Les pointeurs sont utilisés évitant au maximun la duplication des données améliorants ainsi significativement la vitesse de calcul des données. L'utilisation de P-Threads permettant à l'api de dispatcher sont travail sur plusieurs processeurs en suivant le modèle de programmation MapReduce.

Performances

Les performances ont été calculées sur des tâches basiques telles que :

• Compter la fréquence d'un mot dans un texte.
• Trouver les dépendances entre les liens de toutes les pages d'un site web (quel lien relie tel page).
• Multiplication d'une matrice
• Coder un texte avec une clé correspondant à un autre texte (String match).
• KMeans( Classifie des points en 3D dans des groupes)
• La fréquence des couleurs RGB dans un tableau d'images
• Calculer la ligne qui correspond à un nuage de points sur un graphique.

avec comme valeur étalon, les performances des p-threads sans le modèle de MapReduce.

Les résultats de Phoenix montrent qu'avec un processeur 4 coeurs on peux accélérer les calculs de 40% et avec 8 coeurs, on peut aller jusqu'à 50%. Cependant, bien que la vitesse de calcul soit augmentée, sur de simples machines elle reste néanmoins équivalente à celle des p-threads. En effet, bien que le modèle de MapReduce soit très efficace sur des clusters a l'échelle de millions de données. L'implémentation du modèle n'est pas assez générale pour couvrir la totalité des programmes.

Mars

Présentation

Encouragé par le succès du modèle programmation MapReduce, Mars, un framework pour implémenter le modèle MapReduce sur des processeurs graphiques a vu le jour. Les processeurs graphiques (GPU) ont dix fois plus de mémoire que les CPU, et sont aussi dix fois plus rapide.

Performances

Pour évaluer ces performances, Mars a été comparé avec Phoenix, sur les mêmes tâches. Il en ai ressorti que les performances de Mars soit 1.5 fois plus élevés que celles de phoenix.

Optimisation

Mars est encore en cours d'étude, cependant trois points essentiels sont relevés pour une future évolution .

• Mars ne peut pas s'occuper de trop grandes données, il est bloqué à la taille de la mémoire du GPU.
• Mars est implémenté sur des GPU NVIDIA, il serait intéressant de le développer sur les GPUAMD.
• Phoenix est implémenté sur des CPU et Mars sur des GPU. Un framework, combinant les deux API, pourrait voir le jour; dans le but de prendre les avantages des deux types de processeurs.

Alternatives à Hadoop

PQL

Présentation

PQL est un langage implémenté comme une extension de java. Il a été conçu de façon à augmenter son expressivité et sa modularité, une comparaison entre son implémentation et celle de java sur des tâches parallèles similaires a été effectuée. La parallélisation est un problème d’actualité dans l’informatique, c’est pourquoi il est important de créer des langages ou des librairies permettant de faciliter la parallélisation. Cependant il ne faut pas oublier la programmation séquentielle, c’est pourquoi l’implémentation d'un langage adapté à la programmation parallèle et séquentielle est essentiel. Tous les programmes implémentés avec PQL est parallélisé grâce à la programmation déclarative

Programmation Déclarative

PQL est un langage basé sur la programmation déclarative, ce qui veut dire que le programmeur peut spécifier ce qu'il veut faire sans pour autant préciser “comment” le faire. PQL a été implémenté pour trouver la méthode la plus rapide à exécuter. En effet, il va savoir quel bout de code devra être parallélisé ou non. Ce n’est pas au développeur de chercher la façon d’optimiser son code. PQL répartie la charge de travail sur les processeurs disponibles en utilisant le MapReduce. PQL a été optimisé pour les tâches parallèles de façon à ce que l'utilisateur ait le moins de manipulations à faire comparé a un code parallèle écrit par l'utilisateur.

Language illustration

PQL utilise les booléens &&, ||, XAND, XOR ainsi que des mots-clés tel que reduce, forall, query, exists. Le nombre de mots clés de PQL est explicitement réduit pour améliorer un maximum la parallélisation.

Une query peut s’implémenter avec : une Quantifier Expression (renvoie une quantité, utilise les forall, exists …), une Java Expression (du code Java), un id (variable, avec ou sans type), une QExpr (combine les cas précédents).

Exemple :

int[] array = query(Array[x]==y):range(1,1000).contains(x) && y=x*x;

La requête va retourner un tableau d'entiers contenant les carrés de x entre 1 et 1000.

Une quantifier expression peut s’écrire sous la forme : QUANT-EXPR::=<QUANT> <ID> ‘:’ <QUERY> | query‘(’ <MATCH> ‘)’ ‘:’ <QUERY> (Container queries) | reduce‘(’ id ‘)’ <ID>[over<ID-SEQ>]:<QUERY> (Reductions)

Exemple : forall int x : x == x toujours vrai (Quantifier) (ID) (x==x query)

Container queries : query (Map.get(x) == y): range(1, 10).contains(x) && y == x∗x query‘(’ <MATCH> ‘)’ ‘:’ <QUERY> Va construire une map contenant les carrés de 1 à 10 (on peut accéder à l’élément 0 mais ça retournera un null)

Opération de réduction :

reduce (sumInt) x : myset.contains(x)
cette expression va additionner tous les x contenu dans myset ! 

reduce (sumDouble) x over y : set.contains(y) && x == 1.0 / y
Additionne les inverses des éléments contenus dans la map

Les types peuvent être déclarés implicitement (l'utilisateur n'a pas besoin de déclarer le type) ou explicitement (l'utilisateur doit déclarer le type de la variable). PQL lève les exceptions mais on n’a pas de garantie de l’ordre . PQL utilise == ou = pour les égalités (sauf pour les string et les objets .equals()).

L’utilisateur peut créer ses propres réductions. Elle doivent respecter plusieurs priorités (méthode statique, associative, commutative …) Si ces propriétés n'étaient pas respectées, cela entraînerait des erreurs.

Performances

Pour évaluer PQL, différentes tâches ont été implémentées(calcul des primes d’un grand nombre d’employés, trouver une sous-chaîne dans une chaîne de chaînes de caractères, une liste de documents qui peuvent être reliés entre eux, le but étant de trouver les cycles, calculer le nombre d’occurrences de tous les mots dans plusieurs textes).

l'implémentation des solutions suivant les différentes façons :

• Le PQL
• Une méthode Java mono thread
• Multi thread Java
• SQL
• Framework Hadoop

Lors de l’implémentation avec le mono-thread et pql, des variables d’entrées sont utilisées mais sans aucune variable temporaire. Tandis que pour les autres implémentations, l'utilisation de variables et des tableaux intermédiaires était nécessaire que ce soit pour la communication ou l’optimisation. l’implémentation SQL est simple mais deviens très vite compliquée si on la mélange avec du java.

L'implémentation PQL est simple à écrire car elle contient peu de lignes. En effet, la solution qui contient le nombre de lignes minimum est toujours celle avec PQL. La deuxième étant la mono thread (mais non optimisée). Les autres méthodes contiennent 5 à 20 fois plus de lignes.

Dans les 4 tâches, seule la première ne fait pas de bonnes performances car le temps pour faire les opérations de merge des résultats est très élevé. Pour les autres tâches, PQL peut aller entre 2 et 6 fois plus vite que les autres implémentations. Tous ces tests montrent que PQL est bien plus performant et utilise moins de lignes de code. Cependant, il est nécessaire de rappeler que SQL et Hadoop utilise une table de données et donc que leur temps d’exécution est ralenti par l'accès à cette base. De plus, le framework Hadoop est optimisé pour des calculs sur des clusters et non pas sur une seule machine.

Framework "Ecologique"

Présentation

Le MapReduce a surtout été récompensé pour sa flexibilité et son efficacité. Pourtant, à l’heure où l’écologie prend une place plus importante dans les nouvelles technologies, il est pertinent d’étudier la consommation énergétique du MapReduce et savoir comment la diminuer.

La plupart des frameworks implémentant le MapReduce, comme Hadoop, considèrent que le traitement se fait dans un milieu hétérogène. En réalité, il est très rare qu’un cluster contienne les mêmes machines. Les machines varient selon différents critères basés sur la consommation énergétique : quantité de pâte thermique, vitesse du ventilateur, taille du ventilateur, localisé plus ou moins proche d’une unité de refroidissement, quantité de poussière. Cela implique que des machines vont consommer plus d’énergie que d’autres pour réaliser le même traitement.

Un framework a donc été conçu pour pouvoir planifier de façon dynamique les tâches en fonction de la consommation individuelle de chaque nœud du cluster.

Implémentation

Pour pouvoir planifier dynamiquement, il a d’abord fallu élaborer un moyen de connaître la consommation de chaque nœud. Dans cet article , une corrélation entre la température du CPU et la consommation d’énergie. Cependant, il y a des imprécisions dans cette relation. En effet, une machine peut consommer une grande quantité d’énergie tout en ayant une température basse si elle se trouve à proximité d’une unité de refroidissement. Mais, en général, cette corrélation reste correcte.

Lorsque le Nœud Maître du HDFS a distribué les données, il démarre un thread qui va mesurer la température de chaque nœud. Lorsqu’un nœud a fini son traitement, si sa température est inférieure à la température maximale définie par l’utilisateur, alors une nouvelle tâche est affectée au nœud, sinon le nœud sera mis en attente jusqu’à ce que sa température baisse ou que le traitement soit terminé. Toutefois, lorsque toutes les tâches ont été distribuées, le module de tolérance aux pannes va commencer son travail sans prendre en compte la température. En effet, si tous les nœuds sont en refroidissement, la tâche sera alors bloquée.

Pour réduire la consommation, deux paramètres vont être mis en place : Température maximale d’un nœud, nombre de tâches pour réaliser un traitement. Dans les prochaines évolutions de ce framework, chaque nœud aura sa propre température définie car certains processeurs ont une température plus élevée lors de leur fonctionnement.

Performances

Les mesures ont été réalisées sur l’algorithme WordCount (Calcul toutes les itérations de tous les mots d’un document) dans un milieu hétérogène.

Nous allons étudier les variations des performances en fonction des deux paramètres cités dans le chapitre précédent.

La température maximale La limite de température, pour décider si un nœud est capable ou non d’exécuter un traitement, peut limiter les performances. Les tests ont été effectués avec trois températures différentes (80°, 110°, 130°).

Temps de traitement d'un noeud d'un cluster en fonction de sa température

D’après la figure, nous constatons que, plus la température maximale est faible, plus le temps d’exécution est important. Si la température maximale est trop proche de la température des CPU au repos, alors très peu de nœuds obtiendront une nouvelle tâche après avoir compléter la première. Le problème est que cela pourrait entraîner un blocage de la tâche en cours jusqu’à ce qu’un nœud ait une température inférieure à la limite. En récupérant les informations du cluster dans lequel le framework est déployé, on pourrait éviter un tel risque. L’une des informations les plus importantes serait la température de chaque CPU au repos.

De plus, on remarque que plus le nombre de tâches par nœud est important et la température maximale est faible, plus le temps d’exécution augmente. Cela est dû au temps de refroidissement des nœuds pour repasser sous la limite.

Consommation électrique d'un noeud en fonction de la température d'un noeud et de la taille du fichier à traiter

Nous constatons la même tendance pour la consommation électrique de chaque nœud. Avec ce framework, si nous choisissons correctement la température maximale, il est possible de faire des économies significatives sur la consommation électrique tout en limitant la baisse de performance.

Distribution des données La distribution des données dans chaque nœud peut aussi influencer les performances et la consommation du MapReduce. Pour les mesures, nous avons un exemple avec un fichier de 500 Mb et un de 1 Gb.

Temps de traitement d'un noeud en fonction de la température d'un noeud et du partage des données

Dans les 2 graphes, nous constatons 2 points d’intersections, le premier arrive lorsque le ratio tâche/nœud est de 1 (cela signifie qu’il y a autant de tâches que de nœuds). Le deuxième arrive lorsque le ratio est de 2. Cela peut s’expliquer par le fait que le temps d’exécution entre le nœud le plus rapide et le nœud le plus lent est inférieur au temps d’exécution de la tâche. Ce qui implique que chaque nœud a reçu deux tâches. Toutefois, lorsque le ratio est de 3, les tâches sont beaucoup plus courtes, ce qui implique que le temps d’exécution entre le nœud le plus rapide et le nœud le plus lent est supérieur au temps d’exécution de la tâche. Dans la suite des travaux sur ce framework, une méthode sera mise en place pour savoir comment répartir les tâches de façon efficace. Nous pouvons remarquer que la température maximale joue encore un rôle très important dans la vitesse d’exécution.

Pour prouver que la température maximale et la distribution des données influe sur la consommation d’énergie. Des mesures ont été effectuées sur la consommation électrique d’un nœud dans un cluster.

Consommation électrique d'un noeud en fonction de la température d'un noeud et du partage des données

Nous pouvons voir qu’il y a deux piques de consommation pour le fichier de 2 Gb lorsque la température maximale est de 80°. Cela arrive lorsque le nœud est replanifié. Quand le nœud n’est pas replanifié et que les tâches deviennent plus courtes, la division des tâches augmentent et la consommation électrique diminue.

Grâce à ce framework, la consommation électrique d’un nœud a diminué de 36%. Ce framework est encore en cours de développement mais il présente déjà des résultats très intéressants.

Comparaison

Frameworks

	Langage	ratio vitesse clusters	ratio de vitesse processeurs	Economie d'energie	Environnement
Hadoop	Java	1	NC	NC	Cluster
BlobSeer	Java	1.35	NC	NC	Cluster
Mars	Programming GPU	NC	1.1 ~ 4	NC	Processeur graphique
Phoenix	C/C++	NC	1	NC	Processeur multi-coeurs
Framework Ecologique	Java	NC	NC	36%	Cluster

Articles

• « Evaluating MapReduce on Virtual Machines : The Hadoop Case », Cloud Computing 2009, Shadi Ibrahim, Hai Jin, Lu Lu, Li Qi, Song Wu, Xuanhua Shi, 23 novembre 2009

• Dawei Jiang, Beng Chin Ooi, Lei Shi et Sai Wu, « The performance of MapReduce : An In-Depth Study », , vol. 3,‎ 2 septembre 2010, p. 472-483 (ISSN 2150-8097, lire en ligne)

• « Improving MapReduce Performance through Data Placement in Heterogeneous Hadoop Clusters », 24th IEEE International Symposium on Parallel and distributed, Jiong Xie, Shu Yin, Xiaojun Ruan, Zhiyang Ding, Yun Tian, James Majors, Adam Manzanares, Xiao Qin, 23 avril 2010

• « Configuring a MapReduce Framework for Dynamic and Efficient Energy Adaptation », 5th IEEE International Conference on Cloud Computing, Jessica Hartog, Zacharia Fadika, Elif Dede, Madhusudhan Govindaraju, 24 juin 2012

• « Parallel Data Processing with MapReduce : A Survey », ACM, Kyong-Ha Lee, Yoon-Joon Lee, Hyunsik Choi, Yon Dohn Chung, Bongki Moon, 4 décembre 2011

• « BlobSeer: Bringing high throughput under heavy concurrency to Hadoop Map-Reduce applications », 24th IEEE International Symposium on Parallel and distributed, Bogdan Nicolae, Diana Moise,Gabriel Antoniu , Luc Bougé, Mathieu Dorier, 23 avril 2010

• « MapReduce: Simplified Data processing on Large Cluster », ACM, Jeffrey Dean, Sanjay Ghemawat, 2008

• « Evaluating MapReduce for Multi-core and Multiprocesseur Systems », IEEE 13th International Symposium on High Performance Computer Architecture, 2007. HPCA 2007., Colby Ranger, Ramanan raghuraman,Arun Penmetsa,Gary Bradski, Christos Kozyrakis, 2007

• « Improving MapReduce Performance in Heterogeneous Environments », 8th USENIX Symposium on Operating Systems Design and Implementation, Matei Zaharia, Andy Konwinski, Anthony D. Joseph, Randy Katz, Ion Stoica, octobre 2008

• « A Library to Run Evolutionary Algorithms in the Cloud using MapReduce », EvoApplication, Pedro Eazenda , James McDermott, Una-may O’Reilly, 2012

• « Mars : A MapReduce Framework on Graphics Processors », ECOOP 2012 - Object-Oriented Programming - 26th European Conference, Beijing, China, June 11-16, 2012., Bingsheng He, Wenbin Fang, Qiong Luo, Naga K. Govindaraju, Tuyong Wang, 2012

• « PQL: A Purely-Declarative Java Extension for Parallel Programming », 17th International Conference on Parallel Architecture and Compilation Techniques (PACT 2008), Toronto, Ontario, Canada, October 25-29, 2008, Christoph Reichenbach, Yannis Smaragdakis and Neil Immerman, 2008

• « An analysis of traces from a production MapReduce cluster », CMU-PDL-09-107, Soila Kavulya, Jiaqi Tan, Rajeev Gandhi and Priya Narasimhan, 2010

• « A comparison of approaches to large-scale data analysis », Proceedings of the 2009 ACM SIGMOD International Conference on Management of data, A. Pavlo, E. Paulson, A. Rasin, D. J. Abadi, D. J. DeWitt, S. Madden, M. Stonebraker, 2009

• « Mapreduce and parallel dbmss : friends or foss? », Communications of the ACM, M. Stonebraker, D. Abadi, D. J. DeWitt, S. Madden, E.Paulson, A. Pavlo, A. Rasin, 2010

• « Hive - a warehousing solution over a map-reduce framework », Proceedings of the VLDB Endowment VLDB Endowment Hompage archive - August 2009, A. Thusoo, J. S. Sarma, N. Jain, Z. Shao, P. Chakka, S. Anthony, H. Liu, P. Wychoff, R. Murthy, 2009

Articles connexes

• Cloud Computing
• Virtualisation
• Google
• Hadoop
• MapReduce
• Grep