5.1 Mapper uniquement le travail
Il est temps de décrire diverses techniques qui vous permettent d'utiliser efficacement MapReduce pour résoudre des problèmes pratiques, ainsi que de montrer certaines des fonctionnalités de Hadoop qui peuvent simplifier le développement ou accélérer considérablement l'exécution d'une tâche MapReduce sur un cluster.
Comme nous nous en souvenons, MapReduce se compose des étapes Map, Shuffle et Reduce. En règle générale, l'étape Shuffle s'avère être la plus difficile dans les tâches pratiques, car les données sont triées à ce stade. En fait, il existe un certain nombre de tâches dans lesquelles l'étape Carte seule peut être supprimée. Voici des exemples de telles tâches :
- Filtrage des données (par exemple, "Rechercher tous les enregistrements de l'adresse IP 123.123.123.123" dans les journaux du serveur Web) ;
- Transformation des données ("Supprimer la colonne dans les journaux csv");
- Chargement et déchargement de données depuis une source externe ("Insérer tous les enregistrements du journal dans la base de données").
Ces tâches sont résolues à l'aide de Map-Only. Lors de la création d'une tâche Map-Only dans Hadoop, vous devez spécifier un nombre nul de réducteurs :
![](https://cdn.codegym.cc/images/article/392f8537-a932-402b-a637-1c566c5f5c4e/512.jpeg)
Un exemple de configuration de tâche map-only sur hadoop :
interface native | Interface de diffusion Hadoop |
---|---|
Spécifiez zéro nombre de réducteurs lors de la configuration de job'a :
|
Nous ne spécifions pas de réducteur et spécifions un nombre nul de réducteurs. Exemple:
|
Les travaux Map Only peuvent en fait être très utiles. Par exemple, dans la plateforme Facetz.DCA, pour identifier les caractéristiques des utilisateurs par leur comportement, c'est précisément une grande map-only qui est utilisée, dont chaque mappeur prend un utilisateur en entrée et retourne ses caractéristiques en sortie.
5.2 Combiner
Comme je l'ai déjà écrit, l'étape la plus difficile lors de l'exécution d'une tâche Map-Reduce est généralement l'étape de mélange. Cela se produit parce que les résultats intermédiaires (sortie du mappeur) sont écrits sur le disque, triés et transmis sur le réseau. Cependant, il existe des tâches dans lesquelles un tel comportement ne semble pas très raisonnable. Par exemple, dans la même tâche de comptage de mots dans des documents, vous pouvez pré-agréger les résultats des sorties de plusieurs mappeurs sur un nœud map-reduce de la tâche, et transmettre les valeurs déjà additionnées pour chaque machine au réducteur .
![](https://cdn.codegym.cc/images/article/6ca3c77a-f572-4e52-a6a5-2346cce42350/1080.jpeg)
Dans hadoop, pour cela, vous pouvez définir une fonction de combinaison qui traitera la sortie d'une partie des mappeurs. La fonction de combinaison est très similaire à réduire - elle prend la sortie de certains mappeurs en entrée et produit un résultat agrégé pour ces mappeurs, de sorte que le réducteur est souvent également utilisé comme combineur. Une différence importante par rapport à reduce est que toutes les valeurs correspondant à une clé n'atteignent pas la fonction de combinaison .
De plus, hadoop ne garantit pas du tout que la fonction de combinaison sera exécutée pour la sortie du mappeur. Par conséquent, la fonction de combinaison n'est pas toujours applicable, par exemple dans le cas d'une recherche de la valeur médiane par clé. Néanmoins, dans les tâches où la fonction de combinaison est applicable, son utilisation permet d'obtenir une augmentation significative de la vitesse de la tâche MapReduce.
Utilisation du Combiner sur hadoop :
interface native | Diffusion Hadoop |
---|---|
Lors de la configuration de job-a, spécifiez le class-Combiner. En règle générale, c'est la même chose que Reducer :
|
Spécifiez la commande -combiner dans les options de ligne de commande. En règle générale, cette commande est identique à la commande reducer. Exemple:
|
5.3 Chaînes de tâches MapReduce
Il y a des situations où un MapReduce n'est pas suffisant pour résoudre un problème. Par exemple, considérons une tâche WordCount légèrement modifiée : il y a un ensemble de documents texte, vous devez compter combien de mots sont apparus de 1 à 1000 fois dans l'ensemble, combien de mots de 1001 à 2000, combien de 2001 à 3000, et ainsi de suite. Pour la solution, nous avons besoin de 2 tâches MapReduce :
- Nombre de mots modifié, qui pour chaque mot calculera dans lequel des intervalles il est tombé ;
- Un MapReduce qui compte combien de fois chaque intervalle a été rencontré dans la sortie du premier MapReduce.
Solution de pseudo-code :
|
|
|
|
Pour exécuter une séquence de tâches MapReduce sur hadoop, il suffit de spécifier le dossier qui a été spécifié en sortie pour la première en entrée pour la seconde tâche et de les exécuter à tour de rôle.
En pratique, les chaînes de tâches MapReduce peuvent être des séquences assez complexes dans lesquelles les tâches MapReduce peuvent être connectées à la fois séquentiellement et parallèlement les unes aux autres. Pour simplifier la gestion de tels plans d'exécution de tâches, il existe des outils distincts comme oozie et luigi, qui seront abordés dans un article séparé de cette série.
![](https://cdn.codegym.cc/images/article/b14d442d-1528-40b1-a361-676cab6a7a18/800.jpeg)
5.4 Cache distribué
Un mécanisme important dans Hadoop est le cache distribué. Le cache distribué vous permet d'ajouter des fichiers (par exemple, des fichiers texte, des archives, des fichiers jar) à l'environnement dans lequel la tâche MapReduce est en cours d'exécution.
Vous pouvez ajouter des fichiers stockés sur HDFS, des fichiers locaux (locaux à la machine à partir de laquelle la tâche est lancée). J'ai déjà implicitement montré comment utiliser le cache distribué avec le streaming hadoop en ajoutant les fichiers mapper.py et reducer.py via l'option -file. En fait, vous pouvez ajouter non seulement mapper.py et reducer.py, mais des fichiers arbitraires en général, puis les utiliser comme s'ils se trouvaient dans un dossier local.
Utilisation du cache distribué :
API native |
---|
|
Streaming Hadoop |
---|
#nous listons les fichiers qui doivent être ajoutés au cache distribué dans le paramètre –files. L'option --files doit précéder les autres options.
exemple d'utilisation :
|
5.5 Réduire la jointure
Ceux qui sont habitués à travailler avec des bases de données relationnelles utilisent souvent l'opération Join très pratique, qui leur permet de traiter conjointement le contenu de certaines tables en les joignant selon une clé. Lorsque vous travaillez avec des données volumineuses, ce problème se pose également parfois. Considérez l'exemple suivant :
Il existe des journaux de deux serveurs Web, chaque journal ressemble à ceci :
t\t
Exemple d'extrait de journal :
1446792139
178.78.82.1
/sphingosine/unhurrying.css
1446792139
126.31.163.222
/accentually.js
1446792139
154.164.149.83
/pyroacid/unkemptly.jpg
1446792139
202.27.13.181
/Chawia.js
1446792139
67.123.248.174
/morphographical/dismain.css
1446792139
226.74.123.135
/phanerite.php
1446792139
157.109.106.104
/bisonant.css
Il faut calculer pour chaque adresse IP lequel des 2 serveurs elle a visité le plus souvent. Le résultat doit être sous la forme :
\t
Un exemple d'une partie du résultat :
178.78.82.1
first
126.31.163.222
second
154.164.149.83
second
226.74.123.135
first
Malheureusement, contrairement aux bases de données relationnelles, en général, joindre deux logs par clé (dans ce cas, par adresse IP) est une opération assez lourde et se résout à l'aide de 3 MapReduce et du pattern Reduce Join :
![](https://cdn.codegym.cc/images/article/e84f6930-4d46-4b1b-9737-0adcd5994c92/512.jpeg)
ReduceJoin fonctionne comme ceci :
1) Pour chacun des journaux d'entrée, un MapReduce séparé (Map uniquement) est lancé, convertissant les données d'entrée sous la forme suivante :
key -> (type, value
Où key est la clé sur laquelle joindre les tables, Type est le type de la table (premier ou deuxième dans notre cas) et Value est toute donnée supplémentaire liée à la clé.
2) Les sorties des deux MapReduce sont envoyées à l'entrée du 3ème MapReduce, qui, en fait, effectue l'union. Ce MapReduce contient un Mapper vide qui copie simplement l'entrée. Ensuite, shuffle décompose les données en clés et les transmet au réducteur en entrée :
key -> [(type, value)]
Il est important qu'à ce moment le réducteur reçoive les enregistrements des deux journaux, et en même temps, il est possible d'identifier par le champ type de lequel des deux journaux provient une valeur particulière. Il y a donc suffisamment de données pour résoudre le problème initial. Dans notre cas, le réducteur doit simplement calculer pour chaque clé d'enregistrement quel type a rencontré le plus et afficher ce type.
5.6 MapJoins
Le modèle ReduceJoin décrit le cas général de la jointure de deux journaux par clé. Cependant, il existe un cas particulier dans lequel la tâche peut être considérablement simplifiée et accélérée. C'est le cas où l'un des journaux est nettement plus petit que l'autre. Considérez le problème suivant :
Il y a 2 journaux. Le premier journal contient le journal du serveur Web (le même que dans la tâche précédente), le second fichier (taille de 100 Ko) contient l'URL-> Correspondance du thème. Exemple 2ème fichier :
/toyota.php
auto
/football/spartak.html
sport
/cars
auto
/finances/money
business
Pour chaque adresse IP, il faut calculer les pages de quelle catégorie de cette adresse IP ont été chargées le plus souvent.
Dans ce cas, nous devons également joindre 2 logs par URL. Cependant, dans ce cas, nous n'avons pas besoin d'exécuter 3 MapReduce, car le deuxième journal tiendra complètement en mémoire. Afin de résoudre le problème en utilisant le 1er MapReduce, nous pouvons charger le deuxième journal dans le cache distribué, et lorsque le mappeur est initialisé, il suffit de le lire en mémoire, en le mettant dans le -> dictionnaire de sujets.
De plus, le problème est résolu comme suit :
carte:
# find the subject of each of the pages of the first log
input_line -> [ip, topic]
réduire:
Ip -> [topics] -> [ip, most_popular_topic]
Reduce reçoit une adresse IP et une liste de tous les sujets en entrée, il calcule simplement lequel des sujets a été rencontré le plus souvent. Ainsi, la tâche est résolue à l'aide du 1er MapReduce, et la jointure proprement dite a généralement lieu à l'intérieur de la carte (ainsi, si une agrégation supplémentaire par clé n'était pas nécessaire, le travail MapOnly pourrait être supprimé):
![](https://cdn.codegym.cc/images/article/a386bdfc-8d78-4077-83b4-07e560a3b680/512.jpeg)
GO TO FULL VERSION