Configurer un cluster Hadoop multi-nœud

Cet article indique comment configurer un cluster Hadoop à partir d’une configuration en mode pseudo-distribué. Dans un article précédent, on avait expliqué comment installer Hadoop sur Debian GNU/Linux. A la suite de cette installation, le cluster Hadoop ne comportait qu’un seul nœud (i.e. single node cluster) et les jobs MapReduce s’exécutaient de façon pseudo distribuée. De façon à utiliser plus de fonctionnalités d’Hadoop, on va modifier la configuration pour permettre d’exécuter les jobs de façon distribuée.

Dans des articles précédents, on avait expliqué comment installer Debian sur une machine virtuelle VirtualBox pour exploiter la flexibilité des machines virtuelles (VM). On va donc, dans un premier temps, modifier la configuration du nœud qui s’exécutait en mode pseudo distribué pour qu’il soit le name node du cluster. Ensuite, on va dupliquer la machine virtuelle dans laquelle se trouve ce nœud pour créer d’autres nœuds qui seront les data nodes du cluster.

Sommaire

1. Dupliquer les machines virtuelles

2. Configurer le “name node”
Configurer l’alias réseau
Configurer core-site.xml
Configurer hdfs-site.xml
Configurer mapred-site.xml
Configurer yarn-site.xml
Configuration du “master” et des “slaves”
Formattage du système de fichiers HDFS

3. Configurer les “data nodes”
Configuration des alias réseau
Configuration SSH des “data nodes”
Copie de la configuration sur les “data nodes”
Suppression des répertoires de travail Hadoop sur les “data nodes”
Suppression de GNOME sur les “data nodes” (facultatif)

4. Exécuter Hadoop et YARN
Démarrage de Hadoop et YARN
Tester l’exécution
Stopper l’exécution de YARN et Hadoop
Erreur “container is running beyond physical memory limits”

Pour qu’on puisse étudier le fonctionnement d’Hadoop et que le cluster s’exécute en distribuant l’exécution d’un job sur plusieurs nœuds, il faut au moins 2 data nodes (même si un data node suffit à l’exécution). Pour coordonner l’exécution du job, il faut aussi un name node. Il faudra, donc, exécuter simultanément 3 machines virtuelles:

  • 1 machine virtuelle pour le name node
  • 2 machines virtuelles pour les 2 data nodes.

La mise en place de cet environnement va donc consommer beaucoup de mémoire, il faut une machine avec au moins de 12 Go de RAM et quad-core pour permettre aux 3 instances Debian de fonctionner. Ensuite, pour que l’exécution du job soit distribuée à travers le réseau, il faut un routeur et que les machines virtuelles aient des adresses IP différentes pour que les nœuds du cluster puissent communiquer entre eux.

1. Dupliquer les machines virtuelles

La première étape consiste à dupliquer le nœud qu’on avait déjà configuré en mode pseudo-distribué. Mais avant d’effectuer la duplication, on va modifier la configuration pour que chaque machine virtuelle utilise moins de ressources et puisse avoir des adresses IP différentes.

Le système Debian qui se trouve sur la machine virtuelle, doit être arrété en exécutant, par exemple, halt à la ligne de commandes en tant qu’utilisateur root. On peut alors modifier la configuration de la machine virtuelle (VM) dans VirtualBox en faisant un clique droit sur la VM puis en cliquant sur “Configuration”:

  • On peut limiter à un 1 seul processeur en allant dans l’onglet “Système” puis “Processeur”:


  • Modifier la configuration réseau en allant dans l’onglet “Réseau”, en sélectionnant pour la configuration “Mode d’accès réseau” en “Accès par pont”. Ce paramétrage va permettre d’affecter une adresse IP différente pour chaque VM.

  • On peut renommer cette machine “NameNode” en allant dans l’onglet “Général”.

La VM peut être dupliquée en faisant un clique droit puis en cliquant sur “Cloner”. Ensuite il faut cliquer sur “Mode expert” et sélectionner les paramètres suivants:

  • Type de clone: “Clone intégral”
  • Instantanés: “Etat actuel de la machine”
  • Cocher “Réinitialiser l’adresse MAC de toutes les cartes réseau” pour affecter une adresse MAC différente.
  • Nommer la VM, par exemple en “DataNode1”
  • Cliquer ensuite sur “Cloner”.


Il faut ensuite refaire l’opération pour le 2e nœud en nommant la VM “DataNode2”.

A ce stade, on a 3 machines virtuelles: “NameNode”, “DataNode1” et “DataNode2”.

2. Configurer le “name node”

On peut démarrer la VM nommée “NameNode” en cliquant sur “Démarrer”.

Sachant qu’on a commencé avec la VM utilisée après la configuration en mode pseudo distribué à la suite de l’article Installation d’Hadoop sur Debian, certaines étapes ont déjà été effectuées:

  • Java et Hadoop sont installés. Hadoop se trouve dans le répertoire /usr/hadoop/hadoop-2.8.1.
  • L’utilisateur qui peut exécuter Hadoop est “hduser”.
  • Les variables d’environnement JAVA_HOME et HADOOP_HOME sont paramétrées et le répertoire d’Hadoop se trouve dans la variable PATH (cf. Execution de jobs localement)
  • La configuration SSH est effectuée et la clé /home/hduser/.ssh/id_rsa.pub est autorisée car elle a été copiée dans /home/hduser/authorized_keys (cf. Configuration SSH).

Configurer l’alias réseau

Il faut d’abord obtenir l’adresse de la VM en exécutant: 

root@NameNode:~% ip addr show

Dans mon cas, l’adresse IP est 192.168.1.34.

On édite le fichier /etc/hostname en tant qu’utilisateur root (pour se connecter en tant qu’utilsateur root, il faut taper su sur un terminal): 

root@NameNode:~% vi /etc/hostname

On remplace le nom avec le nom “NameNode” (appuyer sur [i] pour passer en mode édition et après avoir effectué le remplacement; pour enregistrer, appuyer sur [Echap] puis taper :wq).

Ensuite il faut éditer le fichier /etc/hosts en tant qu’utilisateur root: 

root@NameNode:~% vi /etc/hosts

On doit commenter toutes les lignes présentes en les précédant du caractère “#” et ajouter la ligne suivante (en remplaçant l’adresse IP avec l’adresse trouvée précédemment): 

192.168.1.34        NameNode

Dans mon cas, le fichier se présente de cette façon: 

#127.0.0.1            localhost 
192.168.1.34       NameNode

On peut redémarrer en exécutant: 

root@NameNode:~% reboot

Configurer core-site.xml

Ce fichier doit être édité en tant qu’utilisateur “hduser”:

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/core-site.xml

Le fichier doit se présenter de cette façon: 

<configuration> 
<property> 
   <name>fs.defaultFS</name> 
   <value>hdfs://NameNode:8020</value> 
   <description>The name of the default file system.</description> 
</property> 
<property> 
   <name>hadoop.tmp.dir</name> 
   <value>/home/hduser/hadoop_data/hd-data/tmp</value> 
</property> 
<property> 
   <name>fs.checkpoint.dir</name> 
   <value>/home/hduser/hadoop_data/hd-data/snn</value> 
</property> 
<property> 
   <name>dfs.data.dir</name> 
   <value>/home/hduser/hadoop_data/hd-data/dn</value> 
</property> 
</configuration>

Dans sa configuration précédente, la valeur de fs.defaultFS était hdfs://localhost:9000. Cette valeur a été remplacée puisque l’alias réseau a été modifié.

Configurer hdfs-site.xml

On édite hdfs-site.xml en tant qu’utilisateur “hduser” de cette façon: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/hdfs-site.xml

Le fichier doit se présenter de cette façon: 

<configuration> 
  <property> 
    <name>dfs.replication</name> 
    <value>2</value> 
  </property> 
  <property> 
      <name>dfs.namenode.name.dir</name> 
      <value>file:/home/hduser/hadoop_data/hdfs/namenode</value> 
  </property> 
  <property> 
      <name>dfs.datanode.data.dir</name> 
      <value>file:/home/hduser/hadoop_data/hdfs/datanode</value> 
  </property> 
  <property> 
      <name>dfs.namenode.checkpoint.dir</name> 
      <value>file:/home/hduser/hadoop_data/hdfs/namesecondary</value> 
  </property> 
</configuration>

Par rapport à la configuration précédente, on a augmenté le paramètre dfs.replication à 2 car on veut configurer 2 data nodes (c’est le facteur de réplication de HDFS) et on a précisé des répertoires de données pour le name node.

Configurer mapred-site.xml

On configure mapred-site.xml en tant qu’utilisateur “hduser”: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/mapred-site.xml

Le fichier doit se présenter de cette façon: 

<configuration> 
<property> 
   <value>yarn</value> 
   <name>mapreduce.framework.name</name> 
</property> 
<property> 
     <name>mapreduce.jobhistory.address</name> 
     <value>NameNode:10020</value> 
</property> 
<property> 
     <name>mapreduce.jobhistory.webapp.address</name> 
     <value>NameNode:19888</value> 
</property> 
<property> 
     <name>yarn.app.mapreduce.am.staging-dir</name> 
     <value>/user/app</value> 
</property> 
<property> 
     <name>mapred.child.java.opts</name> 
     <value>-Djava.security.egd=file:/dev/../dev/urandom</value> 
</property> 
</configuration>

Par rapport à la configuration précédente, on a modifié la valeur du paramètre mapreduce.framework.name en la passant à yarn. Précédemment cette valeur était à local car on exécutait les jobs en mode pseudo-distribué sur un seul nœud. Maintenant YARN doit utiliser les autres nœuds pour l’exécution.

D’autres paramètres ont été rajoutés:

  • mapreduce.jobhistory.address: on indique l’adresse du serveur MapReduce JobHistory.
  • mapreduce.jobhistory.webapp.address: on précise l’adresse web du serveur MapReduce JobHistory.
  • mapred.child.java.opts: ce paramètre permet de rajouter une option au lancement du processus java. Cette option vise à éviter les retards de la JVM dus à la génération de nombres aléatoires. L’attente préalable permettant la génération de “bruit” avant de générer un nombre aléatoire peut bloquer l’exécution du serveur WebLogic SIP utilisé pour exécuter des processus scalables.

Configurer yarn-site.xml

On édite ce fichier en tapant: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/yarn-site.xml

Ce fichier doit se présenter de cette façon: 

<configuration> 
   <property> 
      <name>yarn.nodemanager.aux-services</name> 
      <value>mapreduce_shuffle</value> 
   </property> 
   <property> 
      <name>yarn.resourcemanager.hostname</name> 
      <value>NameNode</value> 
   </property> 
   <property> 
      <name>yarn.resourcemanager.bind-host</name> 
      <value>0.0.0.0</value> 
   </property> 
   <property> 
     <name>yarn.nodemanager.bind-host</name> 
     <value>0.0.0.0</value> 
   </property> 
   <property> 
       <name>yarn.nodemanager.aux-services.mapreduce_shuffle.class</name> 
       <value>org.apache.hadoop.mapred.ShuffleHandler</value> 
   </property> 
   <property> 
       <name>yarn.log-aggregation-enable</name> 
       <value>true</value> 
   </property> 
   <property> 
       <name>yarn.nodemanager.local-dirs</name> 
       <value>file:/home/hduser/hadoop_data/yarn/local</value> 
   </property> 
   <property> 
       <name>yarn.nodemanager.log-dirs</name> 
       <value>file:/home/hduser/hadoop_data/yarn/log</value> 
   </property> 
   <property> 
       <name>yarn.nodemanager.remote-app-log-dir</name> 
       <value>hdfs://NameNode:8020/var/log/hadoop-yarn/apps</value> 
   </property> 
</configuration>

Le détail des éléments de configuration est:

  • yarn.nodemanager.aux-services: on indique le nom du service auxiliaire YARN.
  • yarn.resourcemanager.hostname: on indique le nom réseau du resource manager de YARN.
  • yarn.resourcemanager.bind-host: permet au resource manager d’écouter toutes les interfaces vers lesquelles le serveur RPC et le serveur webapp vont se connecter.
  • yarn.nodemanager.bind-host: ce paramètre est similaire au précédent mais concerne le node manager.
  • yarn.nodemanager.aux-services.mapreduce_shuffle.class: classe Java utilisée pour effectuer l’étape shuffle. Il s’agit de la valeur par défaut.
  • yarn.log-aggregation-enable: on indique qu’on veut activer l’agrégation de logs. Les logs peuvent ainsi être agrégés de façon centrale dans HDFS et non sur le disque de tous les nœuds.
  • yarn.nodemanager.local-dirs et yarn.nodemanager.log-dirs: permettent de préciser des répertoires pour stocker respectivement des données et de logs.
  • yarn.nodemanager.remote-app-log-dir: répertoire dans lequel les logs des node managers seront agrégés. Ce répertoire se trouve dans HDFS.

Configuration du “master” et des “slaves”

On va ensuite indiquer le nom réseau des machines qui seront le name node et les data nodes. On n’a pas encore configuré le nom des alias des data nodes toutefois le nom sera “DataNode1” et “DataNode2”.

On précise l’alias du “NameNode” en éditant le fichier masters: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/masters

On y ajoute seulement le nom du name node: 

NameNode

Ensuite on édite le fichier slaves pour indiquer les alias des data nodes: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/slaves

Le fichier doit contenir les 2 lignes suivantes: 

DataNode1 
DataNode2

Formattage du système de fichiers HDFS

Sachant que le nœud a déjà servi pour des exécutions en mode pseuso-distribué et que la configuration a changé, il faut formatter le système de fichiers HDFS.

Au préalable, pour éviter les erreurs on peut supprimer le contenu des répertoires de travail de HDFS et de MapReduce en exécutant en tant qu’utilisateur “hduser”: 

hduser@NameNode:~% rm –r /home/hduser/hadoop_data/hd-data 
hduser@NameNode:~% rm –r /home/hduser/hadoop_data/hdfs 
hduser@NameNode:~% rm –r /home/hduser/hadoop_data/yarn

On peut, ensuite, formatter en exécutant: 

hduser@NameNode:~% /usr/hadoop/hadoop-2.8.1/bin/hdfs namenode -format

3. Configurer les “data nodes”

Les data nodes doivent, maintenant être configurés. Sachant que ces VM sont issues de duplication de la première VM, tous les éléments nécessaires sont déjà installés. De même, une bonne partie de la configuration a déjà été effectuée. Il y aura juste quelques éléments à modifier.

On peut maintenant démarrer les 2 VM correspondant aux data nodes: la VM “DataNode1” et la VM “DataNode2”.

Configuration des alias réseau

Avant tout, on récupére les adresses IP des 2 VM en exécutant sur les 2 machines la commande suivante: 

hduser#DataNode1:% ip addr show

Et sur l’autre VM: 

hduser@DataNode2:~% ip addr show

Dans mon cas, j’ai les adresses suivantes:

  • DataNode1: 192.168.1.98
  • DataNode2: 192.168.1.50

On indique ensuite le nom réseau pour chaque machine en éditant le fichier /etc/hostname en tant qu’utilisateur root: 

root@DataNode1:~% vi /etc/hostname

Le fichier ne doit contenir que: 

DataNode1

De même pour la VM DataNode2: 

root@DataNode2:~% vi /etc/hostname

Le fichier doit contenir: 

DataNode2

On édite ensuite le fichier /etc/hosts pour indiquer les adresses IP des autres nœuds: 

root@DataNode1:~% vi /etc/hosts

Il faut commenter les autres lignes existantes, ce fichier doit contenir les lignes suivantes (il faut remplacer les adresses IP par celles obtenues précédemment): 

192.168.1.34             NameNode 
192.168.1.89             DataNode1 
192.168.1.50             DataNode2

On effectue la même opération sur la VM DataNode2: 

root@DataNode2:~% vi /etc/hosts
Indiquer les adresses IP des “data nodes” dans le “name node”

Il ne faut pas oublier de répercuter ces modifications aussi dans le name node. On édite donc aussi le fichier /etc/hosts sur la VM “NameNode”: 

root@NameNode:~% vi /etc/hosts

Le fichier doit avoir le même contenu que les autres nœuds: 

192.168.1.34             NameNode 
192.168.1.89             DataNode1 
192.168.1.50             DataNode2

Redémarrage des VM

Il faut redémarrer les VM après cette étape surtout les VM des data nodes “DataNode1” et “DataNode2”.

Configuration SSH des “data nodes”

Sachant que les VM des data nodes sont des clones du name node, la configuration SSH est déjà effectuée. Normalement, il n’y a rien à faire toutefois on précise quelle doit être cette configuration.

La configuration SSH consiste à autoriser le name node à se connecter aux data nodes par connexion SSH. Pour autoriser cette connexion, il faut copier la clé publique du name node dans la liste des clés autorisées dans les data nodes.

La clé publique du name node se trouve dans: /home/hduser/.ssh/id_rsa.pub. Si on regarde son contenu:

hduser@NameNode:~% cat /home/hduser/.ssh/id_rsa.pub 
ssh-rsa AAAAA......p610nf hduser@debianvm

La clé doit être copiée dans le répertoire /home/hduser/.ssh/authorized_keys des 2 data nodes.
On édite ensuite la clé dans la VM “DataNode1” dans /home/hduser/.ssh/authorized_keys pour remplacer hduser@debianvm par hduser@NameNode. La clé doit se présenter de cette façon: 

hduser@DataNode1:~% cat /home/hduser/.ssh/id_rsa.pub 
ssh-rsa AAAAA......p610nf hduser@NameNode

Il faut faire la même modification dans la VM “DataNode2”: 

hduser@DataNode2:~% cat /home/hduser/.ssh/id_rsa.pub 
ssh-rsa AAAAA......p610nf hduser@NameNode

Maintenant l’utilisateur “hduser” du name node peut se connecter sur les 2 data nodes. On peut le vérifier en allant sur le name node et en se connectant sur un terminal en tant qu’utilisateur “hduser”. Si on tape: 

hduser@NameNode:~% ssh DataNode1

On accède directement au DataNode1.

De même, on doit pouvoir se connecter directement sur le DataNode2 en exécutant: 

hduser@NameNode:~% ssh DataNode2

Copie de la configuration sur les “data nodes”

A la suite de la configuration SSH, il faut copier toute la configuration d’Hadoop du name node dans les data nodes en s’aidant de scp.

On se connecte donc, en tant qu’utilisateur “hduser” dans le name node et on exécute les commandes suivantes: 

hduser@NameNode:~% cd /usr/hadoop/hadoop-2.8.1/etc  
hduser@NameNode:~% scp –r hadoop DataNode1:/usr/hadoop/hadoop-2.8.1/etc 
hduser@NameNode:~% scp –r hadoop DataNode2:/usr/hadoop/hadoop-2.8.1/etc

Après cette étape la configuration est dupliquée dans les 2 data nodes.

Suppression des répertoires de travail Hadoop sur les “data nodes”

Sachant que les data nodes proviennent de duplications de machine virtuelle et pour éviter des erreurs, il est préférable de supprimer le contenu des répertoires de travail d’Hadoop en exécutant sur les 2 data nodes les commandes suivantes: 

hduser@DataNode1:~% rm –r /home/hduser/hadoop_data/hd-data 
hduser@DataNode1:~% rm –r /home/hduser/hadoop_data/hdfs 
hduser@DataNode1:~% rm –r /home/hduser/hadoop_data/yarn

Puis sur l’autre data node: 

hduser@DataNode2:~% rm –r /home/hduser/hadoop_data/hd-data 
hduser@DataNode2:~% rm –r /home/hduser/hadoop_data/hdfs 
hduser@DataNode2:~% rm –r /home/hduser/hadoop_data/yarn

Suppression de GNOME sur les “data nodes” (facultatif)

Cette étape vise seulement à diminuer la quantité de mémoire utilisée par la VM. Elle n’est pas indispensable.

Elle consiste à désinstaller GNOME sur les data nodes en exécutant en tant qu’utilisateur root sur chaque data node, la commande suivante: 

root@DataNode1:~% apt-get purge gnome-shell

De même sur l’autre data node: 

root@DataNode2:~% apt-get purge gnome-shell

Il faut redémarrer les 2 data nodes après cette étape.

4. Exécuter Hadoop et YARN

Démarrage de Hadoop et YARN

Après toutes ces étapes, Hadoop peut être démarré en exécutant sur le name node en tant qu’utilisateur “hduser”, les commandes suivantes: 

hduser@NameNode:~% start-dfs.sh

Puis: 

hduser@NameNode:~% start-yarn.sh

Si le démarrage s’effectue correctement, il n’y a pas d’erreurs dans les logs. On peut voir les logs du name node en exécutant la commande suivante: 

hduser@NameNode:~% less /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-namenode-NameNode.log

Les erreurs sont indiquées avec le niveau ERROR.

On peut aussi regarder les logs des data nodes: 

hduser@DataNode1:~% less /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-datanode-DataNode1.log

Et sur l’autre data node: 

hduser@DataNode2:~% less /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-datanode-DataNode2.log

Pour vérifier si tout a démarré normalement, on peut taper sur le name node: 

hduser@NameNode:~% hdfs dfsadmin –report

Normalement le résultat devrait indiquer la présence des 2 data nodes: 

Live datanodes (2): 
  
Name: 192.168.1.50:50010 (DataNode2) 
Hostname: DataNode2 
Decommission Status : Normal 
Configured Capacity: 29455585280 (27.43 GB) 
DFS Used: 114630656 (109.32 MB) 
Non DFS Used: 12166201344 (11.33 GB) 
DFS Remaining: 15654895616 (14.58 GB) 
DFS Used%: 0.39% 
DFS Remaining%: 53.15% 
Configured Cache Capacity: 0 (0 B) 
Cache Used: 0 (0 B) 
Cache Remaining: 0 (0 B) 
Cache Used%: 100.00% 
Cache Remaining%: 0.00% 
Xceivers: 1 
Last contact: Sat Oct 28 12:59:19 CEST 2017 
  
  
Name: 192.168.1.98:50010 (DataNode1) 
Hostname: DataNode1 
Decommission Status : Normal 
Configured Capacity: 29455585280 (27.43 GB) 
DFS Used: 114630656 (109.32 MB) 
Non DFS Used: 12171759616 (11.34 GB) 
DFS Remaining: 15649337344 (14.57 GB) 
DFS Used%: 0.39% 
DFS Remaining%: 53.13% 
Configured Cache Capacity: 0 (0 B) 
Cache Used: 0 (0 B) 
Cache Remaining: 0 (0 B) 
Cache Used%: 100.00% 
Cache Remaining%: 0.00% 
Xceivers: 1 
Last contact: Sat Oct 28 12:59:19 CEST 2017

Tester l’exécution

On peut tester l’exécution avec le job Wordcount, par exemple avec le fichier:
http://www.gutenberg.org/files/2600/2600-0.txt

Ce fichier doit être copié dans HDFS en écrivant: 

hduser@NameNode:~% hdfs dfs -mkdir /user  
hduser@NameNode:~% hdfs dfs -mkdir /user/hduser  
hduser@NameNode:~% hdfs dfs -mkdir /user/hduser/input  
hduser@NameNode:~% hdfs dfs -put 2600-0.txt /user/hduser/input

On peut lancer le job en écrivant: 

hduser@NameNode:~% hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.8.1.jar wordcount /user/hduser/input/2600-0.txt output

En cas de succès, le résultat devrait indiquer: 

Job job_XXXX completed successfully

Stopper l’exécution de YARN et Hadoop

Pour stopper l’exécution, on exécute les commandes suivantes sur le name node: 

hduser@NameNode:~% stop-dfs.sh

Puis: 

hduser@NameNode:~% stop-yarn.sh

Erreur “container is running beyond physical memory limits”

Cette erreur peut se produire lors de l’exécution d’un job MapReduce en particulier avec la version 1.8 de Java.

Ainsi lors de l’exécution des jobs Map, une erreur survient avec le message:

"Container [pid=XXXX,containerID=XXXXX] is running beyond physical memory limits. 
  Current usage: 1.0 GB of 1 GB physical memory used; 2.7 GB of 2.1 GB virtual memory used. 
  Killing container."

Cette erreur empêche au job d’aboutir. La correction de ce problème consiste à limiter la mémoire utilisée par un job en ajoutant certains paramètres dans les fichiers de configuration.

Stopper YARN et Hadoop

Il faut stopper l’exécution de YARN et Hadoop avant d’éditer les fichiers.

En commançant par le name node: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/mapred-site.xml

On ajoute les éléments de configuration suivants qui vont limiter la quantité de mémoire utilisée à 2 Go (la taille est indiquée en Mo): 

<property> 
     <name>mapreduce.map.memory.mb</name> 
     <value>2048</value> 
</property> 
<property> 
     <name>mapreduce.reduce.memory.mb</name> 
     <value>2048</value> 
</property> 
<property> 
     <name>mapreduce.map.java.opts.max.heap</name> 
     <value>2048</value> 
</property> 
<property> 
     <name>mapreduce.reduce.java.opts.max.heap</name> 
     <value>2048</value> 
</property>

Ensuite on édite le fichier yarn-site.xml en exécutant: 

hduser@NameNode:~% vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/yarn-site.xml

On ajoute le paramètre suivant:

<property> 
    <name>yarn.nodemanager.vmem-check-enabled</name> 
    <value>false</value> 
</property>

Ces fichiers doivent être copiés sur les data nodes en exécutant les lignes suivantes: 

hduser@NameNode:~% cd /usr/hadoop/hadoop-2.8.1/etc  
hduser@NameNode:~% scp –r hadoop DataNode1:/usr/hadoop/hadoop-2.8.1/etc 
hduser@NameNode:~% scp –r hadoop DataNode2:/usr/hadoop/hadoop-2.8.1/etc

On peut ensuite démarrer Hadoop et YARN et effectuer un nouveau test d’exécution.

Références
  • How to Setup Hadoop Multi Node Cluster – Step By Step : https://dwbi.org/etl/bigdata/183-setup-hadoop-cluster
  • Avoiding JVM Delays Caused by Random Number Generation: https://docs.oracle.com/cd/E13209_01/wlcp/wlss30/configwlss/jvmrand.html
  • Fichier par défaut mapred-default.xml: https://hadoop.apache.org/docs/r2.7.2/hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml
  • Fichier par défaut yarn-default.xml: https://hadoop.apache.org/docs/r2.4.1/hadoop-yarn/hadoop-yarn-common/yarn-default.xml
  • ERROR:is running beyond physical memory limits: http://community.cloudera.com/t5/Cloudera-Manager-Installation/ERROR-is-running-beyond-physical-memory-limits/td-p/55173
  • Stackoverflow: Container is running beyond virtual memory limits: https://stackoverflow.com/questions/43441437/container-is-running-beyond-virtual-memory-limits
  • Stackoverflow: Container is running beyond memory limits: https://stackoverflow.com/questions/21005643/container-is-running-beyond-memory-limits
  • Tune Hadoop Cluster to get Maximum Performance: http://crazyadmins.com/tag/tuning-yarn-to-get-maximum-performance/

    • Exécuter un job Hadoop MapReduce avec .NET Core

    Cet article fait suite à l’article Hadoop MapReduce en 5 min qui expliquait de façon théorique le mécanisme des jobs MapReduce. Dans ce présent article, le but est de rentrer un peu plus dans les détails de l’implémentation d’un job Hadoop MapReduce avec une technologie .NET.

    Dans un premier temps, on va expliciter les différentes méthodes pour exécuter un job MapReduce. Ensuite on va présenter une méthode pour exécuter un job avec une technologie .NET.

    Sommaire

    Quelques méthodes pour exécuter un job MapReduce

    Préparation de l’environnement de dévelopement
    Installation du SDK .NET Core
    Installation de Visual Studio Code (facultatif)
      Installation de l’extension C# pour Visual Studio Code (facultatif)

    Implémentation d’un job MapReduce avec .NET Core
    Hadoop Streaming
    Implémentation du “mapper”
      Implémentation seulement avec le SDK .Net Core
      Implémentation avec Visual Studio Code
      Tester les assemblies générées
      Exécuter le job MapReduce avec seulement le “mapper”
    Implémentation du “reducer”
      Compiler le “reducer”
      Exécuter le job MapReduce
    Amélioration du “mapper” pour ignorer les majuscules et la ponctuation

    Comparaison des performances entre Hadoop et Hadoop Streaming avec .NET Core

    Pour conclure…

    Quelques méthodes pour exécuter un job MapReduce

    Plusieurs méthodes sont possibles pour exécuter une job MapReduce avec ou sans technologie .NET:

    • Syncfusion Big Data Platform: ce cluster peut être exécuté en local sur Windows et permet d’exécuter des jobs Hadoop MapReduce avec le framework .NET.
    • Microsoft .NET SDK For Hadoop il existait un SDK implémenté par Microsoft pour exécuter un job MapReduce. Malheureusement ce projet n’est officiellement plus supporté depuis janvier 2017.
    • Hortonworks: cet éditeur propose des solutions pour faciliter l’utilisation de l’environnement Hadoop sur la plateforme Cloud de Microsoft avec HDInsight.
    • Ecosystème Hadoop de Cloudera: Cloudera propose une suite complète composée de plusieurs outils Big Data. Cette suite est livrée sous forme de machines virtuelles sur lesquelles on peut accéder aux différents outils. Il est possible d’exécuter des jobs MapReduce sur cette plateforme avec .NET Core.
    • Microsoft Azure avec son produit HDInsight: il permet d’accéder à un grand nombre d’outils Big Data avec des langages très différents dont des langages .NET. Hadoop fait partie des outils proposés par le service HDInsight qui permet de créer facilement des clusters.
    • La plateforme Cloud d’Amazon AWS permet aussi d’accéder à une panoplie d’outils Big Data avec Amazon EMR dont Apache Hadoop.
    • Google Cloud fait aussi partie des plateformes Cloud permettant d’exécuter des jobs MapReduce avec Cloud Dataproc.

    Certaines de ces solutions sont utilisables gratuitement et peuvent suffire à effectuer des tests. Une dernière solution est d’installer directement Hadoop directement sur sa machine ou sur une machine virtuelle. La solution de la machine virtuelle est particulièrement intéressante puisqu’elle donne la possibilité de:

    • Dupliquer des machines virtuelles dans le cas où on veut réutiliser une configuration intéressante,
    • Installer différents outils sans polluer son système d’exploitation principal,
    • Reprendre une installation en créant une nouvelle machine virtuelle et
    • De tenter facilement différentes configurations.

    Enfin l’intérêt d’utiliser directement la suite Apache Hadoop est de bénéficier de sa gratuité.

    Préparation de l’environnement de dévelopement

    Dans le cadre de cet article, on utilisera, une version d’Apache Hadoop installée sur un système Debian GNU/Linux. En préambule, il faut:

    Pour permettre d’exécuter un job Map Reduce avec une technologie .NET sur Linux, on se propose d’utiliser .NET Core. Ainsi, avant d’implémenter le job il faut installer quelques éléments pour exécuter du code C# avec .NET Core.

    Installation du SDK .NET Core

    Le SDK est le seul élément indispensable à installer. Son installation est assez rapide, il suffit de suivre les instructions pour une distribution Debian: https://www.microsoft.com/net/core#linuxdebian.

    Dans le cas du système Debian que l’on a installé sur la machine virtuelle VirtualBox, il faut réaliser les étapes suivantes:

    1. Ouvrir un terminal et se connecter en tant qu’utilisateur root en tapant: 
      su
    2. Installer quelques outils comme curl en exécutant: 
      apt-get install curl libunwind8 gettext apt-transport-https

      curl permettra le téléchargement de la clé publique des produits Microsoft.

    3. Télécharger la clé publique des produits Microsoft en faisant: 
      curl https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor > microsoft.gpg
    4. Déplacer la clé pour qu’elle soit considérée comme une clé de confiance par le système de packages APT: 
      mv microsoft.gpg /etc/apt/trusted.gpg.d/microsoft.gpg
    5. Il faut ensuite éditer le fichier /etc/apt/sources.list.d/dotnetdev.list, par exemple avec vi en exécutant: 
      vi /etc/apt/sources.list.d/dotnetdev.list

      Il faut ensuite aller vers les dernières lignes du fichier, de passer en mode insertion en tapant [i] et ajouter la ligne suivante:

      deb [arch=amd64] https://packages.microsoft.com/repos/microsoft-debian-stretch-prod stretch main

      On peut enregistrer le fichier en faisant [echap] et en tapant la commande suivante:
      :wq puis [entrée]

    6. On met à jour la liste des packages présents dans les différentes sources de APT en tapant: 
      apt-get update
    7. Enfin, on installe le package correspondant au SDK .NET Core en tapant: 
      apt-get install dotnet-sdk-2.0.0

    Installation de Visual Studio Code (facultatif)

    L’installation de Visual Studio Code n’est pas indispensable toutefois Visual Studio Code est un éditeur qui va faciliter l’implémentation du job par la suite.

    Pour installer Visual Studio Code, il suffit de suivre les instructions sur:
    https://code.visualstudio.com/docs/setup/linux.

    Dans notre cas, il faut exécuter les étapes suivantes en tant qu’utilisateur root:

    1. Editer le fichier /etc/apt/sources.list.d/vscode.list en tapant: 
      vi /etc/apt/sources.list.d/vscode.list

      Passer en mode insertion en tapant [i] et ajouter la ligne:

      deb [arch=amd64] https://packages.microsoft.com/repos/vscode stable main

      Il faut enregistrer et quitter en tapant la touche [echap], en écrivant :wq puis [entrée].

    2. On met à jour la liste des packages présents dans les différentes sources de APT en tapant: 
      apt-get update
    3. On installe le package code en tapant: 
      apt-get install code
    4. On peut démarrer Visual Studio Code en tapant: 
      code

    Installation de l’extension C# pour Visual Studio Code (facultatif)

    Cette étape est aussi facultative toutefois l’extension C# facilite grandement l’implémentation.

    Pour installer l’extension C#, il faut:

    1. Cliquer sur View puis Extensions
    2. Taper ensuite: 
      c#
    3. Cliquer ensuite sur “install”.

    A la fin de cette étape, l’extension est prête à être exécutée.

    Implémentation d’un job MapReduce avec .NET Core

    Dans cette partie, on se propose d’implémenter l’exemple classique d’un job MapReduce: Wordcount. Il consiste à compter le nombre de mots dans un texte.

    Cet exemple est très classique toutefois son grand intérêt est d’être facile à mettre en oeuvre car il ne nécessite qu’un fragment de texte sous forme de fichier texte en entrée. Il n’y a, par exemple, pas d’appels à une base de données ou à d’autres services tiers qui pourraient compliquer l’implémentation de ce traitement.

    Brièvement, un job MapReduce est principalement composé de 2 séries d’opérations:

    • Des opérations Map qui permettent de lire les données en entrée et d’appliquer des opérations à toutes les données lues.
    • Une ou plusieurs opérations Reduce qui agrègent les résultats provenant des opérations Map et les écrit.

    Le fonctionnement de MapReduce est expliqué plus en détails dans l’article Hadoop MapReduce en 5 min.

    Hadoop est implémenté en Java et propose un SDK pour ce langage. Dans le cas de .NET Core, il n’y a pas de SDK similaire, il faut donc exécuter le job de façon différente en utilisant Hadoop Streaming.

    Hadoop Streaming

    Hadoop Streaming est une API générique qui permet à Hadoop de s’interfacer avec n’importe quel type de langage.

    On l’a vu précédemment, l’implémentation d’un job MapReduce passe au moins par l’implémentation d’un mapper. Dans la plupart des cas, il faudra aussi implémenter un reducer même si ce n’est pas obligatoire.

    Dans le cas d’Hadoop Streaming, Hadoop s’interface avec le mapper et le reducer en passant par l’entrée et la sortie standard stdin et stout. C’est ce qui permet à Hadoop Streaming de s’interfacer avec n’importe quel type de langage ou de technologie. Dans la pratique, il faut implémenter 2 exécutables correspondant au mapper et au reducer, Hadoop exécute ces exécutables dans des processus différents et leur fournit les données ligne après ligne par l’intermédiaire de l’entrée standard. La sortie des exécutables doit aussi passer par la sortie standard.

    Le mapper doit donc lire le contenu de l’entrée standard par ligne. De la même façon, pour écrire le résultat, le mapper passe par la sortie en utilise la syntaxe suivante:

    • Une ligne représente un paire de clé/valeur
    • La clé et la valeur sont séparées par le caractère tabulation.

    Le reducer fonctionne de la même façon et s’interface avec Hadoop suivant la même méthode:

    • En entrée, il reçoit des paires de clé/valeur séparées par le caractère tabulation.
    • En sortie, il doit fournir les paires de clé/valeur en les séparant par le caractère tabulation.

    Dans la suite, on va donc implémenter un mapper et un reducer du job Wordcount avec .NET Core de façon à produire 2 exécutables utilisables avec Hadoop Streaming.

    Implémentation du “mapper”

    Dans un premier temps, on va implémenter seulement le mapper du job Wordcount en considérant que:

    • Une ligne de l’entrée standard correspond à une ligne du fichier texte en entrée.
    • Chaque ligne en sortie correspond à une paire de clé/valeur et on sépare les 2 avec le caractère tabulation.

    Dans le cas du job Wordcount, le mapper reçoit un ligne de texte et considère les mots comme étant des clés. Il va associer la valeur “1” à chacun des mots. Le résultat produit est donc des paires de clé/valeur avec une clé pour chaque mot et “1” étant la valeur de chaque paire.

    Voici donc l’implémentation du mapper:

    using System; 
using System.Text; 
  
namespace mapper 
{ 
    class Program 
    { 
        static void Main(string[] args) 
        { 
            // Récupération de l'entrée standard 
            var fileLine = Console.ReadLine(); 
 
            // Tant qu'Hadoop fournit des lignes 
            while (fileLine != null)  // il faut tester la nullité et non string.isNullOrEmpty() 
            { 
                var words = fileLine.Split(' '); 
                foreach (var word in words) 
                { 
                    if (string.IsNullOrWhiteSpace(word)) continue; 
   
                    // On fournit le résultat par la sortie standard en séparant 
                    // la clé (le mot) et la valeur 1 avec une tabulation  
                    Console.WriteLine($"{word}\t1"); 
                } 
  
                // Récupération de l'entrée standard 
                fileLine = Console.ReadLine(); 
            } 
        } 
    } 
}

    Implémentation seulement avec le SDK .Net Core

    Pour implémenter sans passer par Visual Studio Code, il faut effectuer les étapes suivantes:

    1. Créer le projet console .NET Core du mapper en exécutant: 
      % dotnet new console -n mapper 
The template "Console Application" was created successfully. 
  
Processing post-creation actions... 
Running 'dotnet restore' on mapper/mapper.csproj... 
  Restoring packages for /home/mat/Documents/mapper/mapper.csproj... 
  Generating MSBuild file /home/mat/Documents/mapper/obj/mapper.csproj.nuget.g.props. 
  Generating MSBuild file /home/mat/Documents/mapper/obj/mapper.csproj.nuget.g.targets. 
  Restore completed in 288.37 ms for /home/mat/Documents/mapper/mapper.csproj. 
 
Restore succeeded.

      Cette ligne produit un répertoire nommé mapper avec les fichiers suivants:

      % ls mapper  
mapper.csproj  obj  Program.cs
    2. Il faut éditer le fichier mapper/Program.cs avec le code plus haut.
    3. On compile en exécutant: 
      % dotnet build     
Microsoft (R) Build Engine version 15.3.409.57025 for .NET Core 
Copyright (C) Microsoft Corporation. All rights reserved. 
  
  mapper -> /home/mat/Documents/mapper/bin/Debug/netcoreapp2.0/mapper.dll 
  
Build succeeded. 
    0 Warning(s) 
    0 Error(s) 
  
Time Elapsed 00:00:03.34
    4. Pour produire l’assembly à déployer, on peut exécuter: 
      % dotnet publish -c "Release" 
Microsoft (R) Build Engine version 15.3.409.57025 for .NET Core 
Copyright (C) Microsoft Corporation. All rights reserved. 
  
  mapper -> /home/mat/Documents/mapper/bin/Release/netcoreapp2.0/mapper.dll 
  mapper -> /home/mat/Documents/mapper/bin/Release/netcoreapp2.0/publish

    L’assembly à exécuter se trouve dans le répertoire mapper/bin/Release/netcoreapp2.0/publish.

    Implémentation avec Visual Studio Code

    Une autre possibilité est d’utiliser Visual Studio Code:

    1. Ouvrir Visual Studio Code en exécutant à la ligne de commandes:
      code
    2. Créer un répertoire vide et l’ouvrir en cliquant sur File ⇒ Open Folder ou avec le raccourci clavier [Ctrl] + [K] et [Ctrl] + [O].
    3. Ouvrir ensuite la ligne de commandes dans Visual Studio Code en cliquant sur View ⇒ Integrated terminal.
      Le terminal se trouve directement dans le répertoire ouvert précédemment.


    4. Dans le terminal de Visual Studio Code, taper: 
      dotnet new console

      Le projet sera directement créer dans le répertoire mapper et les fichiers apparaîtront dans Visual Studio Code:


      Si l’explorateur à gauche n’apparait pas il faut cliquer sur l’icône ou exécuter la raccourci clavier [Ctrl] + [Maj] + [E]:

    5. On peut alors cliquer sur le fichier Program.cs et l’éditer directement en copiant collant l’implémentation précédente.
    6. Un message apparaît dans Visual Studio Code: “Required assets to build and debug are missing for ‘mapper’. Add them ?”

      Il faut répondre [Yes] pour que Visual Studio Code crée un répertoire nommé .vscode qui contiendra des éléments pour compiler et débuguer.

    7. On peut ensuite compiler en effectuant une parmi les opérations suivantes:
      • Cliquer sur Tasks ⇒ Run build tasks
      • Utiliser le raccourci clavier [Ctrl] + [Maj] + [B]
      • Taper dans le terminal Visual Studio Code: dotnet build (il faut enregistrer le fichier avant d’effectuer cette étape car, contrairement aux autres méthodes, l’enregistrement n’est pas automatique).

      A la fin de cette étape, une assembly est produite dans le répertoire mapper/bin/Debug/netcoreapp2.0/.

      Pour débugger…

      On peut à ce moment débugguer directement dans Visual Studio Code en ajoutant un point d’arrêt et en cliquant sur Debug ⇒ Start debugging ou en tapant sur [F5] directement.

    8. On peut ensuite produire l’assembly qui sera déployée en exécutant dans le terminal Visual Studio Code: 
      dotnet publish –c "Release"

      L’assembly produite sera dans le répertoire mapper/bin/Release/netcoreapp2.0/publish.

    Raccourcis clavier

    On peut trouver la liste des raccourcis clavier utilisables sur Visual Studio Code dans sa version Linux dans le document suivant: Visual Studio Code Keyboard shortcuts for Linux.

    Tester les assemblies générées

    On peut tester l’implémentation en simulant une ligne de fichier texte produite par Hadoop:

    % echo "Hadoop est un outil puissant toutefois cet outil a des limites qui le rendent moins puissant" | dotnet bin/Debug/netcoreapp2.0/mapper.dll  
Hadoop    1 
est    1 
un    1 
outil    1 
puissant    1 
toutefois    1 
cet    1 
outil    1 
a    1 
des    1 
limites    1 
qui    1 
le    1 
rendent    1 
moins    1 
puissant    1

    Le résultat produit donc bien une paire de clé/valeur par ligne et pour chaque ligne la clé est séparée de la valeur avec une tabulation.

    Exécuter le job MapReduce avec seulement le “mapper”

    Maintenant on va exécuter le job MapReduce sans reducer. Dans un premier temps, il faut mettre sur HDFS le fichier correspondant à l’entrée du job.
    La manipulation des fichiers sur HDFS peut se faire en utilisant quelques commandes explicitées dans Commandes shell courantes pour HDFS.

    Si on utilise le même fichier que celui utilisé à l’installation de Hadoop (ce fichier est téléchargeable sur http://www.gutenberg.org/files/2600/2600-0.txt) il suffit d’exécuter les commandes suivantes (avec l’utilisateur permettant d’exécuter Hadoop, dans notre cas c’est “hduser”):

    hdfs dfs -mkdir /user  
hdfs dfs -put 2600-0.txt /user

    Ces commandes permettent de mettre le fichier sur HDFS dans le répertoire /user.

    Dans le cas de notre installation, Hadoop a été installé dans le répertoire /usr/hadoop/hadoop-2.8.1/ et donc Hadoop Streaming se trouve dans le répertoire /usr/hadoop/hadoop-2.8.1/share/hadoop/tools/lib/hadoop-streaming-2.8.1.jar.

    On peut donc lancer le job en exécutant:

    % hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/tools/lib/hadoop-streaming-2.8.1.jar \ 
-D mapreduce.job.reduce=0 \ 
-files "bin/Release/netcoreapp2.0/publish" \ 
-mapper "dotnet bin/Release/netcoreapp2.0/publish/mapper.dll" \ 
-input /user/2600-0.txt \ 
-output /output_mapper

    Pour le détail:

    • -D mapreduce.job.reduce=0: permet de préciser qu’on ne veut pas d’étape reduce.
    • -files "bin/Release/netcoreapp2.0/publish": le job va considérer les fichiers du mapper se trouvant dans ce répertoire.
    • -mapper "dotnet bin/Release/netcoreapp2.0/publish/mapper.dll": instruction permettant d’exécuter le mapper.
    • -input /user/2600-0.txt: le fichier en entrée.
    • -output /output_mapper: le répertoire de sortie.

    Si le job s’est correctement exécuté, le contenu du répertoire /output_mapper sur HDFS devrait être:

    % hdfs dfs -ls /output_mapper             
Found 2 items 
-rw-r--r--   1 hduser supergroup          0 2017-10-21 10:49 /output_mapper/_SUCCESS 
-rw-r--r--   1 hduser supergroup    4242981 2017-10-21 10:49 /output_mapper/part-00000

    Pour récupérer le fichier résultat qui se trouve dans le répertoire /output_mapper sur HDFS, il faut exécuter:

    hdfs dfs –get /output_mapper/part-00000

    Si on regarde le contenu du fichier part-00000, on voit donc toutes les clés correspondant aux mots et la valeur 1 séparées par une tabulation.

    Implémentation du “reducer”

    Dans le cas du job Wordcount, il faut implémenter le reducer en considérant que:

    • Chaque ligne en entrée contient une clé et sa valeur séparées par le caractère tabulation
    • Les lignes en entrée sont triées par ordre alphabétique (à cause de l’étape sort).
    • Chaque ligne de sortie doit comprendre une clé correspondant au mot et la valeur qui doit être le nombre d’occurrences trouvées de ce mot. La clé et la valeur sont aussi séparées par une tabulation.

    L’implémentation du reducer est donc:

    using System; 
  
namespace reducer 
{ 
    class Program 
    { 
        static void Main(string[] args) 
        { 
            string currentWord = string.Empty; 
            int currentWordCount = 0; 
            string word = string.Empty; 
  
            // Récupération de l'entrée standard 
            string valueFromMapper = Console.ReadLine(); 
 
            // Tant qu'Hadoop fournit des lignes 
            while (valueFromMapper != null) // il faut tester la nullité et non string.isNullOrEmpty() 
            { 
                // On sépare le mot de la valeur 
                var splitLine = valueFromMapper.Split('\t'); 
                if (splitLine.Length == 2) 
                { 
                    word = splitLine[0]; 
                    int count; 
                    // Tant qu'on obtient le même mot, on augmente le nombre d'occurences 
                    if (Int32.TryParse(splitLine[1], out count)) 
                    { 
                        if (currentWord == word) 
                        { 
                            currentWordCount += count; 
                        } 
                        else 
                        { 
                            // Si le mot change alors on écrit le nombre d'occurrences trouvé pour le mot précédent 
                            if (!string.IsNullOrWhiteSpace(currentWord)) 
                            { 
                                Console.WriteLine($"{currentWord}\t{currentWordCount}"); 
                            } 
                            currentWord = word; 
                            currentWordCount = count; 
                        } 
                    } 
                } 
  
                // Récupération de l'entrée standard 
                valueFromMapper = Console.ReadLine(); 
            } 
  
            // On n'oublie pas d'écrire la ligne correspondant au dernier mot 
            if (currentWord == word && !string.IsNullOrWhiteSpace(currentWord)) 
            { 
                Console.WriteLine($"{currentWord}\t{currentWordCount}"); 
            } 
        } 
    } 
}

    Compiler le “reducer”

    Pour compiler ce code avec .NET Core, on procède comme pour le mapper:

    1. On crée un projet console vide: 
      dotnet new console –n reducer

      Dans Visual Studio Code, il faut fermer le dossier du mapper en cliquant sur File ⇒ Close Folder.
      Puis il faut ouvrir un nouveau dossier: File ⇒ Open Folder pour ouvrir un répertoire vide correspondant à celui du reducer. Dans le terminal de Visual Studio Code, on peut ensuite taper la ligne précédente pour créer un projet console.

    2. On édite le fichier Program.cs avec le code du reducer.
    3. On compile en exécutant: 
      dotnet build

      Le répertoire de sortie de l’assembly est: reducer/bin/Debug/netcoreapp2.0/

    On considère qu’à ce stade, on a 2 répertoires mapper et reducer contenant respectivement l’implémentation du mapper et du reducer:

    % ls 
mapper    reducer

    On peut tester l’implémentation du reducer en exécutant la ligne suivante:

    % echo "Hadoop est un outil puissant toutefois cet outil a des limites qui le rendent moins puissant" | dotnet mapper/bin/Debug/netcoreapp2.0/mapper.dll | sort -k1,1 | dotnet reducer/bin/Debug/netcoreapp2.0/reducer.dll  
a    1 
cet    1 
des    1 
est    1 
Hadoop    1 
le    1 
limites    1 
moins    1 
outil    2 
puissant    2 
qui    1 
rendent    1 
toutefois    1 
un    1

    Enfin pour produire l’assembly du reducer qui servira lors de l’exécution du job MapReduce, il suffit d’exécuter dans le répertoire du reducer la ligne:

    % dotnet publish -c "Release" 
Microsoft (R) Build Engine version 15.3.409.57025 for .NET Core 
Copyright (C) Microsoft Corporation. All rights reserved. 
  
  reducer -> /home/mat/Documents/reducer/bin/Release/netcoreapp2.0/reducer.dll 
  reducer -> /home/mat/Documents/reducer/bin/Release/netcoreapp2.0/publish/

    Le répertoire de sortie de l’assembly du reducer est donc reducer/bin/Release/netcoreapp2.0/publish/.

    Exécuter le job MapReduce

    Avant d’exécuter le job MapReduce, on va copier les fichiers du mapper et du reducer dans un même répertoire:

    % ls 
mapper    reducer 
 
% mkdir publish 
% cp mapper/bin/Release/netcoreapp2.0/publish/* publish 
% cp reducer/bin/Release/netcoreapp2.0/publish/* publish 
% ls publish  
mapper.deps.json  mapper.pdb             reducer.deps.json    reducer.pdb 
mapper.dll      mapper.runtimeconfig.json  reducer.dll    reducer.runtimeconfig.json

    On lance l’exécution en tant qu’utilisateur permettant d’exécuter un job MapReduce (dans notre cas c’est “hduser”) en exécutant la ligne:

    % hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/tools/lib/hadoop-streaming-2.8.1.jar \ 
-files "publish" \ 
-mapper "dotnet publish/mapper.dll" \ 
-reducer "dotnet publish/reducer.dll" \ 
-input /user/2600-0.txt \ 
-output /output_mapreduce

    La différence avec la ligne précédente est qu’on précise un reducer:

    -reducer "dotnet publish/reducer.dll"

    On supprime aussi l’option indiquant de ne pas utiliser d’étape reduce: 

    -D mapreduce.job.reduce=0

    Si le job s’est correctement exécuté, le contenu du répertoire /output_mapreduce sur HDFS devrait être:

    % hdfs dfs -ls /output_mapreduce 
Found 2 items 
-rw-r--r--   1 hduser supergroup          0 2017-10-21 11:44 /output_mapreduce/_SUCCESS 
-rw-r--r--   1 hduser supergroup     487290 2017-10-21 11:44 /output_mapreduce/part-00000

    Pour récupérer le fichier résultat qui se trouve dans le répertoire /output_mapreduce sur HDFS, il faut exécuter:

    hdfs dfs –get /output_reduce/part-00000

    Si on regarde le contenu du fichier, on voit donc les mots et le nombre d’occurrences de ces mots:

    #2600]    1 
$5,000)    1 
($1    1 
(1)    17 
(2)    27 
(3)    12 
(4)    1 
(5)    1 
(801)    1 
(At    1 
(Barclay    1 
(Berg    3 
(Borodinó)    1 
(Borís    1 
(By    1 
(Daniel    1 
(Davout).    1 
(Denísov’s    1 
(Ermólov    1 
(God)    1 
(He    18 
(His    1 
(How    1 
(I    3 
(In    1 
(Instructions    1 
(It    4 
(Joseph    1 
(Karáy    2 
(Kochubéy    1 
(Konovnítsyn    1 
(Kutúzov    1 
(Madame    1 
(Mademoiselle    2 
(Mary    1 
(Moscou,    1 
(Murat)    1 
(Márya    1 
(Natásha    2 
(Ney)    1 
...

    En observant ce fichier, on constate que les caractères de ponctuation et les majuscules gênent le comptage des mots. On se propose, par la suite, une petite amélioration dans le mapper pour éviter que le comptage ne soit perturbé par la ponctuation et les majuscules.

    Amélioration du “mapper” pour ignorer les majuscules et la ponctuation

    Pour éviter de différencier des mots à cause d’un majuscule ou de caractères de ponctuation, on modifie le mapper avec l’implémentation suivante:

    using System; 
using System.Text; 
  
namespace mapper 
{ 
    class Program 
    { 
        static void Main(string[] args) 
        { 
            var fileLine = Console.ReadLine(); 
            while (fileLine != null) 
            { 
                var words = fileLine.Split(' '); 
                foreach (var word in words) 
                { 
                    if (string.IsNullOrWhiteSpace(word)) continue; 
  
                    var wordWithoutPunctuation = StripPunctuation(word.Trim().ToLower()); 
  
                    Console.WriteLine($"{wordWithoutPunctuation}\t1"); 
                } 
  
                fileLine = Console.ReadLine(); 
            } 
        } 
  
        public static string StripPunctuation(string inputString) 
        { 
            var outputString = new StringBuilder(); 
            foreach (char character in inputString) 
            { 
                if (!char.IsPunctuation(character)) 
                    outputString.Append(character); 
            } 
            return outputString.ToString(); 
        } 
    } 
}

    En ré-exécutant le job, on obtient un comptage plus précis:

    ...
a    10494 
aah    1 
ab    1 
aback    3 
abacus    1 
abandon    25 
abandoned    54 
abandoning    26 
abandonment    14 
abandons    1 
abashed    12 
abate    2 
abbreviations    1 
abbé    18 
abbés    1 
abc    1 
abdicate    1 
abdomen    2 
abdomens    2 
abduction    3 
abductors    1 
abhorrence    1 
abide    1 
ability    4 
ablaze    2 
able    107 
abnormal    1 
abnormally    1 
abodes    1 
abolition    1 
abominable    7 
abominably    1 
abounding    1 
about    1016 
abouti    2 
above    146 
aboveboard    1 
abreast    5 
abroad    33 
abrupt    6 
abruptly    10 
abrámovna    1 
absence    48 
absent    4 
absentees    1 
absently    6 
absentminded    14 
absentmindedly    10 
absentmindedness    8 
absolute    16 
absolutely    21 
absolved    1 
absorb    1 
absorbed    32 
absorption    2 
abstain    2 
abstained    2 
abstaining    2 
abstemious    1 
...

    Une autre amélioration pourrait être de ne pas prendre en compte les nombres lors du comptage.

    Comparaison des performances entre Hadoop et Hadoop Streaming avec .NET Core

    Pour terminer, on peut essayer de comparer la vitesse d’exécution du job MapReduce entre:

    • Hadoop avec l’implémentation en Java et
    • Hadoop Streaming avec une implémentation du mapper et reducer en .Net Core.

    Etant donné que l’exécution du job avec les données précédentes est rapide, on va dupliquer ce fichier pour augmenter les données d’entrée. On effectue alors une trentaine de copies du même fichier et on place ces fichiers dans HDFS:

    % ls comparison  
2600-0.txt   2600-12.txt  2600-15.txt  2600-18.txt  2600-20.txt  2600-23.txt  2600-26.txt  2600-29.txt    2600-3.txt  2600-6.txt    2600-9.txt 
2600-10.txt  2600-13.txt  2600-16.txt  2600-19.txt  2600-21.txt  2600-24.txt  2600-27.txt  2600-2.txt    2600-4.txt  2600-7.txt 
2600-11.txt  2600-14.txt  2600-17.txt  2600-1.txt   2600-22.txt  2600-25.txt  2600-28.txt  2600-30.txt    2600-5.txt  2600-8.txt 
  
% hdfs dfs -put comparison /user/

    On exécute ensuite le job wordcount en Java sur le répertoire:

    % hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.8.1.jar wordcount /user/comparison/ output_java

    On effectue la même exécution avec Hadoop Streaming et .NET Core sans l’amélioration précédente:

    % hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/tools/lib/hadoop-streaming-2.8.1.jar -files "publish" -mapper "dotnet publish/mapper.dll" -reducer "dotnet publish/reducer.dll" -input /user/comparison/ -output /output_dotnetcore

    L’exécution en Java a pris 56 secondes et l’exécution avec Hadoop Streaming s’est déroulée en 138 secondes dans les mêmes conditions. Ce qui est important dans ce test, ce n’est pas le temps d’exécution mais la différence entre l’exécution en Java et celle utilisant Hadoop Streaming.

    Pour conclure…

    En conclusion, cet article a permis de montrer comment utiliser .NET Core pour implémenter un mapper et un reducer de façon à exécuter un job Hadoop MapReduce. Avec Hadoop Streaming, l’implémentation n’est pas forcément plus complexe même si l’interfaçage en utilisant la sortie et l’entrée standard nécessite une adaptation du code.
    Le gros inconvénient de Hadoop Streaming est la pénalité imposée sur les performances puisque passer par l’entrée et la sortie standard engendre un lenteur plus importante qu’une implémentation avec un SDK.
    Toutefois Hadoop Streaming permet d’exploiter d’autres fonctionnalités de Hadoop pour optimiser l’exécution des jobs comme l’utilisation d’un cache distribué ou d’un combiner. Ces fonctionnalités feront l’objet d’articles futurs.

    Références

    Hadoop MapReduce en 5 min

    Hadoop (i.e. High-availability distributed object-oriented platform) est une système distribué permettant de stocker et d’analyser des données. Le grand intérêt d’Hadoop est de proposer un framework pour effectuer des analyses de données de façon parallélisée sur plusieurs machines. D’autre part, Hadoop permet d’utiliser des machines normales et de les associer en groupe de façon à ce que les puissances de calcul soient additionnées. Ces groupes de machines sont appelés des clusters et chaque machine est appelée un nœud (ou node). En effet, pour effectuer de gros traitements il n’est pas nécessaire d’utiliser de gros serveurs de type “mainframe”, plusieurs PC individuels de puissance classique suffisent.

    Les avantages principaux d’Hadoop sont:

    • Permet une grande scalabilité pour stocker et traiter de grandes quantités de données.
    • Est tolérant aux pannes.
    • Est optimisé pour un type varié de données (comme le texte ou les images).
    • Possède un écosystème riche et souvent gratuit composé de différents outils capables de résoudre une problématique précise.

    Les traitements effectués par Hadoop sont appelés des jobs, ils sont exécutés suivant un paradigme unique appelé MapReduce. Ce paradigme a pour but de rendre l’implémentation d’un job moins complexe pour le développeur en l’affranchissant de toutes les problématiques liées au calcul distribué.

    Ainsi un job MapReduce va effectuer plusieurs opérations:

    • Séparer les données en entrée en plusieurs morceaux indépendants (appelés chunks) qui pourront être traités en parallèle. Les chunks en entrée sont traités par des tâches appelées “map tasks”.
    • La sortie des tâches ou “map tasks” est ensuite triée.
    • Les données d’entrée et de sortie d’un job MapReduce sont stockées sur un système de fichiers distribués appelé HDFS (i.e. Hadoop distributed file system).
    • Le framework Hadoop permet de programmer l’exécution des tâches suivant la puissance de calcul, de monitorer l’exécution des tâches et de ré-exécuter les tâches dont l’exécution aurait échouée.
    • Un job MapReduce peut être implémenté sans effectuer de développement relatif à la parallélisation des tâches, à la distribution du travail sur les différentes machines du système ou des opérations de récupération des données sur le système de fichiers distribués. Il n’y a pas non plus de changements de l’implémentation vis-à-vis de l’adaptation du cluster à la charge.

    Le but de cet article est d’expliciter les grandes lignes de l’architecture d’Hadoop, des différents éléments qui composent ce système et d’expliquer brièvement comment implémenter un job MapReduce.
    Cette article permet d’avoir quelques connaissances théoriques pour permettre d’exécuter des jobs Hadoop MapReduce. Il n’y a pas d’explications de code mais juste des explications du point de vue algorithmique. Le code permettant d’exécuter un job en .NET sera explicité, par la suite, dans un article futur.

    Sommaire

    L’écosystème Hadoop
    HDFS
      Name node
      Secondary name node
      Data node
    YARN
      Resource Manager
      Node manager
      Application master
      Container
    MapReduce
      Architecture d’Hadoop

    Explication générale d’un job MapReduce
    Map
    Reduce
      Shuffle
      Sort
      Reduce
    Exemples de jobs MapReduce
      Word count
      Moteur de recherche
      Multiplication d’une matrice et d’un vecteur

    En conclusion

    L’écosystème Hadoop

    Comme indiqué plus haut, l’écosystème Hadoop est très riche et est composé plusieurs outils qui peuvent être organisé en couches. Les composants dans une même couche ne communiquent pas entre eux, ils communiquent avec des composants sur les couches supérieures ou inférieures:

    • Hadoop HDFS (i.e. Hadoop Distributed File System)
    • Yarn (i.e. Yet Another Resource negotiator): gestion de ressources.
    • MapReduce: modèle de programmation pour permettre d’appliquer un calcul (Map) et d’agréger les résultats (Reduce).
    • Hive et Pig: modèle de programmation haut niveau au dessus de MapReduce pour apporter une abstraction: Hive permet d’effectuer des requètes SQL et Pig permet d’effectuer du scripting sur des flux de données.
    • Giraph: permet d’analyser des graphes pour les réseaux sociaux.
    • Storm, Spark, Flink: traitement en mémoire et en temps réel.
    • HBase, Cassandra, MongoDB: base de données NoSQL permettant de gérer des collections de clé-valeurs ou de grandes tables séparées (i.e. “sparse tables”).
    • Zookeeper: permet de centraliser la configuration, la synchronisation et la disponibilité de ces outils.


    Dans l’installation de base d’Hadoop, seuls les composants HDFS, Yarn et MapReduce sont présents. Les autres composants doivent etre installés de façon séparée.
    Tous ces outils sont libres et peuvent être utilisés ensemble.

    Des solutions existent pour livrer l’ecosystème entier avec tous les outils sans considérer des installations séparées comme Cloudera, MapR ou Hortonworks.
    Ces solutions doivent être installées sur un système hôte ou sur des machines virtuelles. Il est aussi possible d’utiliser un écosystème entier avec des solutions Cloud comme Amazon EC2, HDInsight de Microsoft Azure ou Cloud Dataproc de Google Cloud.

    Pour effectuer des tests, on peut aussi facilement installer Hadoop sur une Debian GNU/Linux en utilisant une machine virtuelle.

    HDFS

    HDFS (i.e. Hadoop Distributed File System) est un système de fichiers dont la caractéristique principale est de permettre à Hadoop d’utiliser des fichiers en entrée pour l’exécution de ses jobs et d’écrire le résultat des jobs de façon distribuée. La manipulation des fichiers devient ainsi complètement transparente.

    Hadoop est un système distribué et nécessite de pouvoir traiter des fichiers sur plusieurs nœuds d’un cluster. HDFS est capable de répondre aux exigences posées par Hadoop suivant plusieurs critères:

    • HDFS peut répartir des fichiers sur plusieurs machines en additionnant la taille de chaque nœud de façon à ne former qu’un seul espace de stockage.
    • HDFS répartit plusieurs fractions d’un fichier sur plusieurs nœuds ce qui permet de stocker des fichiers dont la taille dépasse la taille du disque d’un nœud. Les fractions de fichier ont des tailles fixes de 128 MO, ces fractions s’appellent des “chunks”.
    • Pour éviter les pertes de données dans le cas de crash d’un disque, HDFS réplique ses fichiers sur plusieurs nœuds d’un cluster. Il existe aussi des paramètres pour que HDFS prenne en compte les racks dans lesquels sont rangés les serveurs. Ainsi dans la réplication, il peut isoler encore davantage les données.
    • L’écriture et la manipulation des données par HDFS sont transparentes. Le système de fichiers apparaît comme un espace de stockage unique alors qu’il est composé de l’addition de plusieurs disques sur plusieurs nœuds du cluster.

    Pour éviter un trop grand nombre d’échanges de données entre nœuds d’un cluster, Hadoop minimise les données qui doivent transiter d’un nœud à l’autre lors du traitement d’un job. Les traitements se font, en priorité, sur le nœud où se trouve les données. Le programme effectuant les traitements est déplacé sur le nœud où il sera exécuté car il est généralement beaucoup moins volumineux que les données qu’il manipule.

    HDFS permet, en outre, de stocker de grandes quantités de données dans de larges datasets de façon scalable c’est-à-dire qu’il est possible d’augmenter l’espace disque en ajoutant des nœuds au cluster. HDFS distingue plusieurs types de nœuds: les name nodes et les data nodes.

    Chaque fichier stocké dans HDFS possède des méta-données qui sont:

    • Le nom du fichier,
    • Les données de permission relatives à ce fichier: similaire aux permissions dans un système Unix.
    • La localisation des blocs relatifs à ce fichier sur les data nodes.

    La taille des blocs de fichiers étant assez grande (128 MO par défaut), ce qui permet de réduire le nombre de blocs utilisés pour stocker un ficher. Moins le nombre de blocs est important et moins nombreux seront les méta-données associés à chaque fichier. D’autre part, les blocs de grande taille permettent de favoriser la lecture en continu des données sur un disque. Plus les données seront lues de façon continue sur les disques et plus un job MapReduce sera exécuté de façon efficace et rapide. A l’opposé, la lecture de fichiers de taille faible implique des accés répétés et succints à des disques pouvant se trouver sur des nœuds différents. Ce qui tend à diminuer les performances.

    La répartition des données est uniforme sur les nœuds du cluster. Un composant appelé le balancer s’exécute régulièrement pour répartir les données sur les nœuds de façon uniforme dès que la différence est trop importante. Si on ajoute un nœud au cluster, le balancer s’exécute aussi pour répartir les données sur le nouveau nœud lors d’une phase de réallocation de blocs. Supprimer un nœuds d’un cluster correspond aussi à une opération particulière.

    Sur le schéma suivant, on peut voir qu’il existe un name node par cluster. Les clients ajoutent ou suppriment des fichiers de façon transparente directement sur les data nodes en passant par HDFS. Pour exécuter un job, les clients s’adressent au name node qui sollicite les data nodes. La réplication se fait directement entre data nodes, il n’y a pas d’échanges de données entre le name node et les data nodes.

    Sources: https://hadoop.apache.org/docs/r1.2.1/hdfs_design.html

    On peut facilement requêter HDFS à l’aide d’instructions à la ligne commandes.

    Name node

    Il s’agit d’un service central (appelé aussi master node). Il a pour but de:

    • Gérer l’état du système de fichiers
    • Maintient l’arborescence du système de fichiers d’Hadoop et les méta-données associées aux fichiers. Les méta-données permettent d’indiquer où le fichier se trouve physiquement.
    • Le name node connaît les autres types de nœuds appelés data nodes où se trouvent les blocs permettant de stocker les fichiers.
    • Quand le client intervient sur le système de fichiers, il le fait sur les data nodes. Le name node effectue ensuite les demandes aux data nodes pour répondre aux requêtes du client.

    Les méta-données associées à chaque fichier sont stockées sur le disque auprès du name node dans des fichiers edits_xxx et fsimage_xxx (se trouvant dans le répertoire de données de HDFS: <répertoire de données>/hd-data/tmp/dfs/name/current).

    Les positions des blocs de fichier dans les data nodes sont stockées en mémoire (de façon à permettre un accès rapide) et sont reconstruites à chaque démarrage du name node (pendant une phase appelée “safe mode”). Ainsi plus le nombre de fichiers stockée dans HDFS est important et plus il y aura de positions stockées dans la mémoire du name node. L’utilisation de fichiers de grande taille permet de minimiser la quantité de meta-données stockée dans la mémoire du name node. Enfin la mémoire du name node peut être une limite s’il y a trop de meta-données à stocker.

    Secondary name node

    Ce nœud copie les méta-données se trouvant auprès du name node et vérifie l’état de fonctionnement du name node. Dans le cas où ce dernier présente une défaillance, le secondary name node est capable de prendre le relais en prenant la place du name node principal.

    Le passage du secondary name node au name node n’est pas automatique. Il nécessite une opération manuelle.

    Data node

    Les fichiers sont stockés sur les nœuds de ce type. Ils assurent une fonction de stockage et permettent au name node d’accéder aux données quand une tâche le nécessite.

    Quand un fichier est ajouté à HDFS, il est séparé en plusieurs blocs de taille fixe lors d’une étape de “partitionning” et chaque bloc est ensuite répliqué sur plusieurs data nodes du cluster (1 cluster est lui-même composé de plusieurs data nodes). Par défaut, les blocs mémoire font 128 MO et le facteur de réplication de ces blocs est de 3, toutefois ces valeurs sont configurables.

    Les data nodes répondent aux différentes demandes du name node et utilise les méta-données fournies par le name node pour:

    • Créer des blocs
    • Supprimer des blocs
    • Répliquer des blocs

    Le name node ne fait qu’indiquer des traitements à effectuer aux data nodes, il ne lit pas directement les données. Lors de copies de données, par exemple, les copies sont effectuées entre data nodes, elles ne transitent pas par le name node.

    Le partitionning permet des accès parallèles aux blocs du fichier. La réplication réponds à plusieurs objectifs:

    • Tolérance aux pannes: s’il y a un défaut d’un nœud ou une coupure réseau, les données ne sont pas perdues et sont facilement retrouvables.
    • Performance: il est plus rapide d’exécuter un traitement sur des données si elles se trouvent déjà sur le nœud qui va effectuer le traitement.

    L’inconvénient de la réplication est qu’elle nécessite plus d’espace de stockage.

    YARN

    YARN (i.e. Yet Another Resource Negotiator) est un composant qui se trouve au dessus de HDFS qui permet de gêrer l’exécution de jobs en parallèle. Il sert de base et est utilisé par d’autres applications comme MapReduce, Spark etc…

    Comme pour HDFS, YARN est composé de plusieurs services ayant des fonctions spécifiques: le resource manager, le node manager, l’application master et le container.

    Echanges entre les services lors d’une job MapReduce

    D’une façon générale, YARN organise l’exécution d’un job MapReduce dans un cluster en minimisant le déplacement des données et l’utilisation de ressources pour que les performances soient maximisées. Les données nécessaires à l’exécution d’un job ne se trouvent pas forcément dans le nœud où l’exécution sera effectuée. Ainsi, le traitement se fait en priorité où les données se trouvent pour augmenter la bande passante entre les clusters.

    Resource Manager

    Ce service permet d’ordonnancer et de dispatcher les tâches sur le cluster Hadoop. Quand YARN reçoit un job à exécuter, c’est le resource manager qui reçoit le job et qui le dispatche le plus rapidement possible sur les autres nœuds du cluster. Le cluster doit être exploité au maximum de ses capacités lors de l’exécution du job. Quand le resource manager reçoit le job sous forme de package MapReduce, il crée un application manager (on verra par la suite la fonction de ce composant) qu’il va transmettre à un nœud pour qu’il puisse exécuter le job.

    Pour que le resource manager puisse assurer son rôle de coordination des ressources du cluster, il reçoit pendant l’exécution du job le statut des nœuds et l’application master lui transmet des requêtes pour obtenir de la ressource. C’est le resource manager qui va assigner à l’application master les ressources nécessaires à l’exécution du job.

    Node manager

    Chaque nœud ou machine du cluster dispose d’un service appelé node manager qui va organiser les resources de la machine. Le node manager envoie périodiquement le statut du nœud au resource manager pour que ce dernier puisse répartir la charge efficacement suivant les ressources.

    Application master

    Cette application est créée par le resource manager pour exécuter un job. En fonction des statuts que les node managers transmettent au resource manager, ce dernier va affecter l’exécution de l’application master auprès d’un nœud. Pendant l’exécution du job, le resource manager reçoit les informations de statut du job grâce au node manager.

    L’application master a pour but de négocier les ressources auprès du resource manager. Quand le job est terminé, l’application master indique au resource manager la fin de son exécution. Le nœud est alors désalloué par le resource manager pour qu’il soit de nouveau disponible pour un nouveau job.

    Container

    Un container est une ressource abstraite correspondant à une machine: cette notion regroupe la mémoire, le réseau et d’autres ressources de la machine. Les containers envoient à l’application master, les statuts d’exécution d’un job MapReduce.

    MapReduce

    MapReduce est à la fois un paradigme de développement de jobs pour Hadoop et il est aussi un composant d’Hadoop. Le composant Hadoop MapReduce permet d’organiser et d’exécuter des jobs en parallèle sur un système de fichiers distribués HDFS.

    Plusieurs autres outils de l’écosystème Hadoop utilisent Hadoop MapReduce comme par exemple:

    • Hive qui propose une infrastructure de stockage et une syntaxe SQL pour la requêter.
    • Pig qui est un langage de flux de données et un framework d’exécution de job en parallèle.

    L’exécution de jobs en parallèle nécessite la maitrise de notions complexes comme les sémaphores, la mémoire partagée entre threads, de locks etc… Hadoop MapReduce permet d’éviter de prendre en compte toutes ces notions pour se focaliser davantage sur l’implémentation du job au sens fonctionnel à condition de respecter le concept de programmation MapReduce. Ce paradigme permet à Hadoop d’exécuter ses tâches de façon isolée pour optimiser la scalabilité d’un cluster.

    A cause du paradigme de programmation MapReduce, tous les traitements ne sont pas forcément possibles. Ainsi pour utiliser MapReduce, il faut que:

    • Les calculs soient parallélisables: si certaines opérations dépendent l’une de l’autre, il n’y aura pas d’intérêt à utiliser Hadoop et un job MapReduce.
    • Traiter de grands volumes de données: la mise en place des tâches d’exécution de l’opération Map peuvent prendre à elles-seules 1 min. Donc pour que l’exécution d’un job MapReduce soit efficace, il faut que le traitement soit effectué sur un grand nombre de données.
    • Le résultat final soit plus petit que l’ensemble de données de départ: dans le cas contraire, un job MapReduce n’est pas forcément pertinent. En effet un part du job consiste à agréger des valeurs donc s’il n’y a pas d’agrégation, une part de l’exécution de ce job est inutile.
    • Etre capable de convertir les données en liste de paire clé/valeur: cette contrainte peut être difficile à mettre en place mais est primordiale pour l’exécution du job MapReduce car la clé est la seule façon d’avoir un élément permettant, d’abord de ségréger différentes valeurs pour permettre un traitement parallélisable (durant les étapes Map) et ensuite la clé est encore utilisée pour agréger les valeurs (durant la ou les étapes Reduce).

    Architecture d’Hadoop

    Avant d’expliquer l’exécution d’un job MapReduce et maintenant que certaines notions ont été définies, on peut expliquer l’architecture d’un cluster Hadoop.

    Comme on a pu le voir plus haut, Hadoop s’occupe de récupérer les données pour effectuer les traitements distribués, de distribuer les tâches et de gérer les erreurs. Un cluster Hadoop nécessite plusieurs nœuds pour être complètement fonctionnel:

    • Master nodes: il en faut 2, le premier en mode actif et l’autre en mode passif. Le 2e remplace le premier en cas de défaillance. Le master node contient les services suivants:
      • Le name node HDFS
      • Le resource manager YARN
    • Worker nodes: il en faut 3 au minimum. Les données sont ainsi répliquées 3 fois sur les 3 nœuds. On peut rajouter d’autres nœuds, plus le nombre de nœuds est important et plus les tâches s’exécuteront rapidement. Comme indiqué plus haut, les données ne se trouvent pas forcément dans le nœud dans lequel les jobs vont s’exécuter. YARN effectue le transfert d’un worker node à l’autre en minimisant les transferts de données pour augmenter les performances. Chaque worker node contient les services:
      • Le data node HDFS
      • Le node manager YARN

    Synthèse des services principaux exécutés sur les nœuds

    Le master node effectue la coordination et les worker nodes s’occupent du stockage et du traitement des données. Il n’est pas nécessaire d’avoir une configuration hardware particulière pour les worker nodes.

    En cas d’échecs, plusieurs points de redondance sont assurés pour palier à une panne d’un master node ou d’un worker node:

    • Si un worker node est en échec: les données sont répliquées 3 fois donc en cas d’échec d’un worker node, le cluster est encore fonctionnel et les données ne sont pas perdues. Un mécanisme de heartbeat permet à YARN de savoir si un worker node est hors-ligne ou non. Si un worker node devient absent ou si seulement un disque d’un nœud crashe, les données sont copiées sur un autre worker node pour être sûr d’avoir 3 réplications des données.
    • Si le master node est en échec: on ne peut plus utiliser YARN donc le nœud doit être mis en mode passif et le 2e master node qui était en mode passif doit être mis en mode actif. Le mécanisme de passage du mode passif à actif n’est pas automatique, il doit se faire manuellement.

    Explication générale d’un job MapReduce

    MapReduce est basé sur des concepts de programmation fonctionnelle:

    • Des opérations Map: permet de lire les données en entrée et d’appliquer des opérations à toutes les données lues.
    • Une ou plusieurs opérations Reduce: qui agrège les résultats provenant des opérations Map et les écrit.

    Map

    Les opérations Map sont effectuées par un composant logiciel appelé le mapper. Concrètement il s’agit d’une classe dans laquelle il faut implémenter les fonctions qui effectueront le traitement suivant:

    • Récupérer des données d’entrée fournies par Hadoop (les données d’entrée font partie des paramètres d’entrée du job Hadoop). Dans le cas d’un fichier texte, c’est Hadoop qui lit le fichier et non le mapper. Les données d’entrée du mapper sont directement les lignes du fichier sous forme de chaînes de caractères.
    • Filtrer ces données si nécessaire,
    • Ranger les données sous forme de paires clé/valeur et
    • Renvoyer la liste de paires clé/valeur dans Hadoop.

    Ainsi une instance du mapper s’exécute dans une tâche et sur un bloc de données. Différentes instances du mapper s’exécutent en parallèle. Le nombre de tâches Map dépend du nombre de données en entrée toutefois le bon niveau de parallélisme est atteint entre 10 et 100 tâches Map par nœud. A la fin de son exécution, le mapper produit des données intermédiaires qui ne sont pas le résultat final. Il renvoie ces données à Hadoop qui effectuera d’autres traitements.

    C’est dans l’étape Map et dans l’implémentation du mapper qu’il faut choisir la granularité des données qui seront traitées par l’étape Map. Par exemple, dans le cas où on doit effectuer des opérations sur les lignes d’un fichier, c’est cette étape qui indiquera si on doit lire un mot à la fois, une ligne à la fois ou une lettre à la fois etc…
    Si on décide de lire une ligne à la fois, une tâche consistant à effectuer un traitement sur une ligne ne doit pas avoir de liens avec la ligne précédente ou la ligne suivante. Sachant que les traitements sont faits en parallèle, ils ne peuvent pas dépendre l’un de l’autre.

    Par défaut, les données d’entrée du mapper sont des lignes (TextInputFormat) mais il est possible de modifier ce format d’entrée en précisant un autre format d’entrée (KeyValueInputFormat, FileInputFormat ou SequenceFileInputFormat). Préciser un autre format d’entrée permet de changer un autre niveau de granularité des données. Par exemple à ce niveau, on pourrait traiter les fichiers un à un ou par paragraphe au lieu de les traiter par lignes.

    Reduce

    L’étape Reduce est composée de 3 sous-étapes: Shuffle, Sort et une étape Reduce à proprement parlé.

    Shuffle

    Cette étape est effectuée par Hadoop. Elle permet de récupérer les paires de clé/valeur obtenues à la suite de l’opération Map puis de les mixer et de les trier en les répartissant dans des shuffle nodes. Les shuffle nodes sont, en fait, les mêmes nœuds que ceux utilisés lors de l’étape Map toutefois lorsque les paires de clé/valeur sont obtenues, elles sont réaffectées à des nœuds pouvant être différents de ceux dont elles sont issues. L’étape de Shuffle effectue, ainsi, une réaffectation des paires clé/valeur en fonction des clés.

    Les étapes de Shuffle commencent dès qu’une opération Map s’est terminée et elles s’exécutent en parallèle. Les étape Shuffle s’exécutent en même temps que les étapes Sort.

    Sort

    L’étape de Sort est aussi effectuée par Hadoop. Elle s’exécute de façon simultanée avec l’étape Shuffle. Elle consiste à trier les paires clé/valeur et à les répartir dans les shuffle nodes suivant leur clé. Ainsi les paires possédant la même clé seront rangées en priorité dans le même shuffle node.

    Cette étape permet de préparer l’étape Reduce de façon à réduire sa charge de travail. Comme indiqué précédemment, l’étape de Sort est effectuée en même temps que l’étape Shuffle.

    Etape shuffle et sort lors de l’exécution d’un job MapReduce

    Reduce

    Cette étape est effectuée par un composant logiciel appelé reducer. De la même façon que pour le mapper, le reducer est une classe dans laquelle des fonctions effectuent un traitement fonctionnel pour traiter les paires de clé/valeur de façon à les agréger.

    En effet, après l’étape Sort, les données d’entrée de l’étape Reduce sont des listes de paires clé/valeur regroupées par clé. Le reducer devra donc:

    • Récupérer les paires de clé/valeur, ces paires sont fournies par Hadoop,
    • Agréger les valeurs ayant la même clé
    • Renvoyer la liste de paire clé/valeur réduite à Hadoop pour qu’il écrive sur le disque le résultat du job.

    Sachant que l’étape de Reduce a consisté à agréger les valeurs ayant la même clé, le résultat comprendra moins de paires que le nombre de paires fournies en entrée. A l’issue de l’étape Reduce, on obtiendra une liste de paires avec la clé et des valeurs agrégées.

    Par défaut, il n’y a qu’une seule étape Reduce pour le job même s’il y a plusieurs nœuds. Dans le cas d’une seule étape Reduce, il y aura un seul fichier résultat contenant toutes les valeurs agrégées.

    Exemple de l’utilisation d’un seul reducer lors d’un job MapReduce

    Par configuration, il est possible d’indiquer que l’on souhaite plusieurs étapes Reduce. Dans ce cas, les opérations Reduce seront effectuées en parallèle et il y aura autant de fichier résultat que d’opération Reduce. Même s’il y aura plusieurs fichiers contenant les résultats, avoir plusieurs étapes Reduce permet de paralléliser cette étape de façon à réduire son temps d’exécution.

    Exemple de l’utilisation de plusieurs reducers lors d’un job MapReduce

    A l’opposé, on peut ne pas effectuer d’étape de Reduce. Si elle n’est pas effectuée, le fichier résultat contiendra une liste de paire clé/valeur sans agrégation des valeurs.

    Exemples de jobs MapReduce

    Word count

    Le job “word count” est l’exemple classique utilisé pour illustrer un job MapReduce. Il consiste à compter le nombre de mots dans un texte. Ce job est facile à mettre en oeuvre car il ne nécessite qu’un fragment de texte sous forme de fichier texte en entrée. Il n’y a, par exemple, pas d’appels à une base de données ou à d’autres services tiers qui pourraient compliquer l’implémentation de ce traitement.

    D’autre part, le job “word count” se prête bien à l’exécution d’un job MapReduce car si on considère chaque ligne comme étant un élément d’entrée, on peut facilement paralléliser le comptage.

    En entrée, on a donc un texte sous forme de fichier. Ce texte comporte un certain nombre de lignes, elles-mêmes composées de plusieurs mots. Hadoop va découper chaque ligne en blocs séparés qui donneront des entrées différentes pour les tâches Map.

    Les étapes du job MapReduce sont les suivantes:

    • Map: chaque tâche Map reçoit une ligne. La fonction implémentant l’étape Map va considérer les mots comme étant des clés et elle va associer la valeur "1" à chacun des mots. On obtient donc des paires de clé/valeur avec un clé pour chaque mot et "1" étant la valeur de chaque paire.

      Ainsi si on considère la ligne:
      “Hadoop est un outil puissant toutefois cet outil a des limites qui le rendent moins puissant”

      A l’issue de l’étape Map, les paires clé/valeur obtenues sont:

      ("Hadoop"/1), ("est"/1), ("un"/1), ("outil"/1), ("puissant"/1), 
("toutefois"/1), ("cet"/1), ("outil"/1), ("a"/1), ("des"/1), ("limites"/1),
("qui"/1), ("le"/1), ("rendent"/1), ("moins"/1), ("puissant"/1)

      On peut remarquer que les mots "outil" et "puissant" sont répétés et chaque occurrence donne une paire différente.

      Lors de l’étape de Map, on peut filtrer les valeurs, par exemple, en ne gardant que les mots de plus de 3 caractères. Après avoir filtré et mis toutes les lettres en minuscule, la liste de paires devient:

      ("hadoop"/1), ("outil"/1), ("puissant"/1), ("toutefois"/1), 
("outil"/1), ("limites"/1), ("rendent"/1), ("moins"/1), ("puissant"/1).
    • Shuffle et Sort: à l’issue de cette étape, les paires sont regroupées suivant leur clé et réparties sur les différents nœuds du cluster.
      Par exemple, on peut avoir la répartition:

      • Node 1: ("hadoop"/1), ("limites"/1), ("moins"/1), ("outil"/1), ("outil"/1).
      • Node 2: ("puissant"/1), ("puissant"/1), ("rendent"/1), ("toutefois"/1).
    • Reduce: cette étape va agréger les valeurs suivant leur clé. Cette agrégation se fait en additionnant les valeurs ayant la même clé. On obtient donc après la phase Reduce: 
      ("hadoop"/1), ("limites"/1), ("moins"/1), ("outil"/2), ("puissant"/2), 
("rendent"/1), ("toutefois"/1).

    Le fichier de sortie du job contient donc la liste des mots avec le nombre d’occurrences.

    Moteur de recherche

    Ce job est un version simplifiée des traitements effectuées par un moteur de recherche. Elle consiste à associer un mot à des adresses URL. Lorsqu’une page est analysée, il en ressort des mots clés associés à des URL. De manière à ce que chaque mot puisse diriger vers les pages où ils apparaissent, il faut effectuer un traitement pour associer un mot à une liste d’URL.

    Comme on l’a dit, l’étape d’analyse consiste à détecter des mots clés dans une page. Le résultat de ce traitement est une liste de mots avec une URL correspondant à la page où ce mot apparait.
    L’exécution d’un job MapReduce permettrait d’associer un mot à une liste d’URL de façon à exploiter cette liste de paires. Les étapes du job sont:

    • Map: on considère les mots comme étant des clés d’une paire mot/url. L’étape Map va consister à extraire la liste de ces paires à partir d’une opération d’analyse des pages. Cette étape permet aussi de filtrer les mots non pertinents comme les conjonctions de coordination, les verbes trop usuels etc…
      A la suite de cette étape, on a donc une liste de paires mot/url, par exemple:
      (mot1/url1), (mot2/url2), (mot3/url3), (mot2/url1), (mot3/url4), (mot3/url5), (mot1/url6), (mot3/url1), (mot1/url1), (mot1/url6), (mot1/url7).

      On remarque que cette liste contient des doublons, que certains mots dirigent vers des URL différentes ou que des mots différents dirigent vers la même URL.

    • Shuffle et Sort: cette étape va consister à regrouper les paires par mot clé et à les répartir suivant les nœuds du cluster. Par exemple, on peut obtenir la répartition suivante:
      • Node 1: (mot1/url1), (mot1/url6), (mot1/url1), (mot1/url6), (mot1/url7)
      • Node 2: (mot2/url2), (mot2/url1), (mot3/url3), (mot3/url4), (mot3/url5), (mot3/url1)
    • Reduce: cette étape consiste à agréger chaque URL au mot auquel elle est associée. Pour chaque mot, on obtient une liste d’URL. Les doublons peuvent servir à associer une pertinence aux résultats de façon à les trier. A la suite de cette étape, on obtient pour chaque mot:
      • mot1: url1 (pertinence 2), url6 (pertinence 2), url7 (pertinence 1).
      • mot2: url2 (pert. 1), url1 (pert. 1).
      • mot3: url3 (pert. 1), url4 (pert. 1), url5 (pert. 1), url1 (pert. 1).

    Ainsi quand on écrit un mot, on peut voir l’ensemble des URL qui y sont associées classées par pertinence.

    Multiplication d’une matrice et d’un vecteur

    Cette exemple est légèrement plus complexe car il faut savoir comment multiplier une matrice à un vecteur.

    Si on considère une matrice de taille 3×3:

    Et un vecteur de taille 3:

    L’opération de multiplication d’une matrice et d’un vecteur consiste à additionner chaque ligne de la matrice aux valeurs du vecteur ce qui donne trois opérations. Dans notre cas:

    • 1ère opération: 2*6 + 8*3 + 0*5 = 36
    • 2è operation: 0*6 + 4*3 + 1*5 = 17
    • 3e opération: 5*6 + 0*3 + 3*5 = 45

    Le résultat de l’opération est le vecteur:

    Pour effectuer ce calcul avec un job MapReduce, on va considérer un triplet (i, j, aij) avec i le numéro de ligne, j le numéro de colonne et aij la valeur. Si la valeur dans la matrice est 0, elle ne sera pas représentée.

    Dans cet exemple, on effectue le calcul pour une matrice de petite taille toutefois, pour justifier l’utilisation d’un job MapReduce et pour que ce calcul soit pertinent, il faudrait une opération dont la matrice est de taille très grande qui nécessiterait plusieurs minutes de calcul.

    Dans le cas de notre exemple, les opérations du job MapReduce sont:

    • Map: dans un premier temps, on considère que les valeurs du vecteur sont représentables sous la forme (i, vi) et que les valeurs de la matrice sont aij (pour la valeur située à la i-ème ligne et j-ème colonne). L’opération Map consiste à extraire une paire (i, aij * vi) qui consiste à effectuer la multiplication de la valeur aij (située à la ligne i et à la colonne j de la matrice) avec la valeur du vecteur vi (située à la ligne i du vecteur).

      Sachant que le résultat est un vecteur, on peut représenter le résultat de l’opération sous la forme de la paire indiquée précédemment avec i étant le numéro de ligne comme clé de la paire. La valeur étant le résultat de la multiplication aij * vi.

      Il y a donc, autant de tâche Map que de multiplication à effectuer. Dans notre cas, les paires obtenues sont:
      Pour la 1ère ligne (0 est l’indice de la 1ère ligne):

      • 1ère opération Map: (0, 2 * 6) soit (0,12)
      • 2e opération Map: (0, 8 * 3) soit (0, 24)
      • Il n’y a pas de 3è opération pour la 1ère ligne car la valeur du vecteur est 0.

      Pour la 2e ligne (1 est l’indice de la 2e ligne):

      • 3e opération Map: (1, 4 * 3) soit (1, 12). Comme précédemment à cause du 0, on passe directement au calcul de la 2è valeur de la ligne.
      • 4e opération Map: (1, 1 * 5) soit (1, 5)

      Pour la 3e ligne (2 est l’indice de la 2e ligne):

      • 5e opération Map: (2, 5 * 6) soit (2, 30)
      • 6e opération Map: (2, 3 * 5) soit (2, 15)

      A l’issue de toutes les opérations Map, on obtient la liste de paires: (0,12), (0, 24), (1, 12), (1, 5), (2, 30) et (2, 15).
      Cette liste est indiquée de façon ordonnée toutefois dans la réalité elle n’est pas ordonnée car les opérations Map se font de façon parallèle et les résultats ne sont pas forcément ordonnancés de cette façon.

    • Shuffle et Sort: cette étape va ordonner les paires suivant leur clé et répartir équitablement les paires sur les nœuds. On pourrait avoir une répartition de cette façon:
      • Node 1: (0,12), (0, 24), (1, 12), (1, 5)
      • Node 2: (2, 30) et (2, 15)
    • Reduce: cette étape va permettre d’agréger toutes les valeurs suivant leur clé. On aura donc la liste de valeurs:
      • (0, [12, 24])
      • (1, [12, 5])
      • (2, [30, 15])

      En additionnant les valeurs agrégées, on obtient la liste: (0, 36), (1, 17) et (2, 45). Sachant que les clés correspondent à l’indice de ligne du vecteur, on a bien le résultat de la multiplication:

    En conclusion

    Le but de cet article était d’expliquer l’architecture d’Hadoop et le fonctionnement d’un job MapReduce. Il n’est pas rentré dans le détail de l’implémentation d’un mapper ou d’un reducer car ces composants feront l’objet d’un article futur sur le sujet. Cet article sert de base théorique de façon à aborder un job MapReduce de façon plus approfondie par la suite.

    Références

    Commandes shell courantes pour HDFS

    Quelques commandes courantes pour HDFS

    Pour ces commandes, il existe 2 syntaxes possibles:

    • Avec hadoop: avec une syntaxe du type hadoop fs <commande>,
    • Avec hdfs: la syntaxe est hdfs dfs <commande>.

    Cette commande sont proche de celles utilisées par le shell linux comme ls, mkdir, rm, cat, etc…

    Pour lister le contenu d’un répertoire

    hdfs dfs -ls <chemin du répertoire>

    Par exemple:

    hdfs dfs -ls /
hdfs dfs -ls /user  # pour voir le contenu du répertoire "user"
Found 2 items
-rw-r--r--   1 hduser supergroup    3324334 2017-09-16 12:00 /user/135-0.txt
-rw-r--r--   1 hduser supergroup    3359550 2017-09-16 12:01 /user/2600-0.txt

    On peut utiliser aussi:

    hadoop fs -ls /user

    Pour afficher le contenu d’un fichier

    
hdfs dfs -cat <chemin du fichier>

    Par exemple:

    hdfs dfs -cat /user/135-0.txt

    On peut utiliser:

    hadoop fs -cat /user/135-0.txt

    Pour créer un répertoire

    hdfs dfs -mkdir <chemin du nouveau répertoire>

    Par exemple:

    hdfs dfs -mkdir /user/output

    Pour copier un fichier sur HDFS

    On peut utiliser:

    hdfs dfs -put <chemin du fichier source> <chemin du fichier destination sur HDFS>

    La commande suivante est réservé seulement au fichier locaux:

    hdfs dfs -copyFromLocal  <chemin du fichier source> <chemin du fichier destination sur HDFS>

    Par exemple:

    hdfs dfs -put TextFile.txt /user

    ou

    hdfs dfs -copyFromLocal TextFile.txt /user

    Les syntaxes équivalentes avec hadoop sont possibles:

    hadoop fs -put <chemin du fichier source> <chemin du fichier destination sur HDFS>
hadoop fs -copyFromLocal <chemin du fichier source> <chemin du fichier destination sur HDFS>

    Pour effectuer un copie de fichier

    hdfs dfs -cp <chemin du fichier source sur HDFS> <chemin du fichier destination sur HDFS>

    Par exemple:

    hdfs dfs -cp /user/TextFile.txt /user/output
hdfs dfs -cp /user/TextFile.txt /user/TestFile2.txt

    Avec hadoop:

    hadoop fs -cp /user/TextFile.txt /user/output
hadoop fs -cp /user/TextFile.txt /user/TestFile2.txt

    Pour récupérer un fichier sur HDFS

    hdfs dfs -get <chemin du fichier sur HDFS> <chemin du fichier en local>

    Par exemple:

    hdfs dfs -get /user/TextFile2.txt 
hdfs dfs -get /user/TextFile2.txt LocalTextFile2.txt

    Cette syntaxe est réservée aux fichiers locaux:

    hdfs dfs -copyToLocal /user/TextFile2.txt

    ou

    hadoop fs -get /user/TextFile2.txt
hadoop fs -copyToLocal /user/TextFile2.txt

    Les mêmes syntaxes existent pour effectuer des déplacements:

    • hdfs dfs -moveToLocal pour déplacer de HDFS vers le volume local
    • hdfs dfs -moveFromLocal pour déplacer du volume local vers HDFS
    • hdfs dfs -mv pour effectuer des déplacements dans HDFS

    Pour supprimer un fichier

    hdfs dfs -rm <chemin du fichier sur HDFS>

    Par exemple:

    hdfs dfs -rm /user/TextFile2.txt
Deleted /user/TextFile2.txt

    ou

    hadoop fs -rm /user/TextFile2.txt

    Pour supprimer un répertoire

    Si le répertoire est vide, on peut utiliser comme sur le shell rmdir:

    hdfs dfs -rmdir <chemin du répertoire vide>

    Par exemple:

    hdfs dfs -rmdir /user/output2

    Si le répertoire contient des fichiers:

    hdfs dfs -rm -r <chemin du répertoire>

    Par exemple:

    hdfs dfs -rm -r /user/output

    Avec hadoop:

    hadoop fs -rmdir /user/output2
hadoop fs -rm -r /user/output

    Installation d’Hadoop sur Debian

    Cet article explique l’installation de Hadoop sur Debian GNU/linux 9. La version d’Hadoop utilisée est celle téléchargeable directement du site d’Apache.
    L’intérêt d’utiliser Debian est que beaucoup d’outils sont disponibles sur cette distribution et qu’elle est gratuite.
    Dans cet article, on détaille l’installation d’Hadoop et de YARN ainsi que la configuration d’un cluster en single-node.

    Une version de Debian installée sur une machine virtuelle est une configuration très flexible pour essayer Hadoop sans polluer son système d’exploitation hôte. L’installation de Debian sur VirtualBox est facile à réaliser (cf. Installation de Debian sur une machine virtuelle VirtualBox). Toutefois cette étape n’est pas obligatoire car il est possible d’installer Hadoop directement sur plusieurs types de système d’exploitation.

    Sommaire

    1. Installation JDK
    3. Préparation de l’installation d’Hadoop
    Installation SSH
    Configuration SSH
    4. Configuration d’Hadoop en single node cluster
    Configuration d’Hadoop
      Configuration du cluster en mode pseudo-distribué
      Changement des droits sur le répertoire d’Hadoop
    Execution de jobs localement
    Erreur “Connection refused”
    5. YARN
    Configuration de YARN
    Exécution de YARN
    6. Execution d’un job MapReduce
    7. Stopper l’exécution de YARN et Hadoop

    1. Installation JDK

    Avant d’installer Hadoop, il faut installer Java si ce n’est pas déjà fait.
    On peut télécharger la JDK sur le site d’Oracle: http://www.oracle.com/technetwork/java/javase/downloads/index.html.

    La version utilisée était la jdk1.8.0_144 pour linux en 64 bits.

    On peut installer la JDK sur le disque en exécutant:

    tar zxvf jdk-8u144-linux-x64.tar.gz

    Cette commande permet d’extraire les fichiers contenus dans l’archive en listant tous les fichiers extraits.

    En tant qu’utilisateur root, placer le répertoire résultant dans /usr/java. Pour passer en tant qu’utilisateur root, il faut taper:

    su

    Puis pour copier le répertoire:

    mkdir /usr/java
mv <nom du répertoire> /usr/java

    2. Télécharger Hadoop sur le site d’Apache

    On peut télécharger Hadoop sur le site d’Apache: http://hadoop.apache.org/.

    La version à installer est la dernière version stable sur http://hadoop.apache.org/releases.html.
    Dans notre cas la version était la 2.8.1.

    Il faut extraire l’archive et la placer dans /etc/hadoop:

    tar zxvf hadoop-2.8.1.tar.gz

    Et:

    mkdir /etc/hadoop
mv <nom du répertoire> /usr/hadoop

    3. Préparation de l’installation d’Hadoop

    Création d’un groupe et d’un utilisateur Hadoop

    Cette étape n’est pas indispensable mais permet de séparer les permissions. Cet utilisateur sera spécifique à Hadoop.
    Pour créer le groupe “hadoop” et l’utilisateur “hduser”, il faut exécuter en tant qu’utilisateur root (taper su pour switcher vers l’utilisateur root):

    addgroup hadoop
adduser --ingroup hadoop hduser

    On peut ensuite se connecter avec cet utilisateur en écrivant:

    su hduser

    Installation SSH

    Il faut installer SSH en tapant en tant qu’utilisateur root:

    apt-get install ssh

    Il faut aussi installer rsync qui permet des synchronisations de fichiers à distance en utilisant SSH. L’installation se fait en écrivant:

    apt-get install rsync

    Configuration SSH

    L’utilisation d’Hadoop se fait par l’intermédiaire d’une connexion SSH même si Hadoop est exécuté localement. Il faut autoriser l’utilisateur “hduser” à se connecter en SSH sur la machine en créant une clé et en autorisant cette clé sur la machine.

    La création de la clé se fait en exécutant les commandes suivantes. Il faut être connecté en tant qu’utilisateur “hduser”:

    ssh-keygen -t rsa -P ""

    L’option -t indique qu’on souhaite créer une clé de type rsa. L’option -P sert à indiquer une passphrase qui dans ce cas est vide.

    La clé est rangée dans le répertoire /home/hduser/.ssh/id_rsa.pub.

    Il faut ajouter cette clé aux clef autorisées pour les connexions SSH en exécutant:

    cat /home/hduser/.ssh/id_rsa.pub >> /home/hduser/authorized_keys

    Cette commande va rajouter le contenu de id_rsa.pub dans le fichier authorized_keys en le créant s’il n’existe pas.

    On peut ensuite tester l’accès en SSH en écrivant:

    ssh localhost

    Il faut ensuite accepter la connexion en écrivant “yes”. L’accès doit se faire sans passphrase puisqu’on a indiqué une passphrase vide plus haut.

    On peut sortir de la connexion SSH en écrivant:

    exit

    4. Configuration d’Hadoop en single-node cluster

    L’installation peut se faire en suivant les instructions sur: Hadoop: Setting up a Single Node Cluster.

    Cette installation permet de configurer Hadoop pour que le cluster ne soit former que d’un seul nœud.

    Configuration d’Hadoop

    Dans un premier temps, il faut configurer le chemin de la JDK dans le fichier de configuration d’Hadoop hadoop-env.sh. Ce fichier se trouve dans /usr/hadoop/hadoop-2.8.1/etc/hadoop/hadoop-env.sh. On peut l’éditer en tant qu’utilisateur root et en remplaçant:

    export JAVA_HOME=${JAVA_HOME}

    par:

    export JAVA_HOME=/usr/java/jdk1.8.0_144/

    On peut ensuite exécuter Hadoop en tapant:

    /usr/hadoop/hadoop-2.8.1/bin/hadoop

    L’exécution devrait afficher l’aide concernant Hadoop:

    Usage: hadoop [--config confdir] [COMMAND | CLASSNAME]
  CLASSNAME            run the class named CLASSNAME
 or
  where COMMAND is one of:
  fs                   run a generic filesystem user client
  version              print the version
  jar <jar>            run a jar file
                       note: please use "yarn jar" to launch
                             YARN applications, not this command.
  checknative [-a|-h]  check native hadoop and compression libraries availability
  distcp <srcurl> <desturl> copy file or directories recursively
  archive -archiveName NAME -p <parent path> <src>* <dest> create a hadoop archive
  classpath            prints the class path needed to get the
                       Hadoop jar and the required libraries
  credential           interact with credential providers
  daemonlog            get/set the log level for each daemon
  trace                view and modify Hadoop tracing settings

Most commands print help when invoked w/o parameters.

    Configuration du cluster en mode pseudo-distribué

    Cette configuration correspond à la partie “Pseudo-Distributed operation”. Elle permet de configurer le cluster en mode pseudo-distribué (puisqu’il n’y a qu’une seule machine).

    Il faut éditer le fichier core-site.xml dans le répertoire: /usr/hadoop/hadoop-2.8.1/etc/hadoop/core-site.xml en ajoutant:

    <configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>

    Cette configuration permet d’indiquer que le nom du système de fichier est fs.defaultFS et que les répertoires et fichiers dans HDFS sont préfixés avec: hdfs://localhost:9000.

    On édite ensuite le fichier qui se trouve dans: /usr/hadoop/hadoop-2.8.1/etc/hadoop/hdfs-site.xml et on ajoute les éléments suivants:

    <configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

    Cette configuration permet d’indiquer le nombre de réplication d’un bloc qui sera de 1 car il n’y a qu’un seul nœud dans le cluster.

    Changement des droits sur le répertoire d’Hadoop

    Pour permettre l’exécution d’Hadoop par l’utilisateur “hduser”, il faut changer les droits sur le répertoire d’Hadoop en tapant:

    chown -R hduser:hadoop /usr/hadoop/hadoop-2.8.1

    Execution de jobs localement

    En se connectant en tant qu’utilisateur “hduser” (en tapant su hduser), on peut formater le système de fichiers de HDFS en tapant:

    /usr/hadoop/hadoop-2.8.1/bin/hdfs namenode -format

    On peut créer le répertoire qui va contenir les logs lors de l’exécution en écrivant:

    mkdir /usr/hadoop/hadoop-2.8.1/logs

    On peut paramétrer les variables d’environnement pour ajouter aux chemins le répertoire d’Hadoop. Il faut éditer le fichier /home/hduser/.bashrc en ajoutant:

    export JAVA_HOME=/usr/java/jdk1.8.0_144/
export HADOOP_HOME=/usr/hadoop/hadoop-2.8.1/
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin

    Pour prendre en compte les modifications des variables d’environnement, on peut redémarrer sa session de terminal.

    On peut démarrer ensuite le daemon du NameNode et du datanode en exécutant:

    start-dfs.sh

    Sachant que le répertoire contenant ce fichier a été rajouté dans la variable d’environnement PATH, on peut exécuter cette action de n’importe où. Toutefois si ce n’est pas le cas, ce fichier se trouve dans /usr/hadoop/hadoop-2.8.1/sbin/start-dfs.sh.

    Les logs correspondant à cette action sont dans:

    /usr/hadoop/hadoop-2.8.1/logs

    L’exécution devrait indiquer:

    Starting namenodes on [localhost]
localhost: starting namenode, logging to /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-namenode-debianvm.out
localhost: starting datanode, logging to /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-datanode-debianvm.out
Starting secondary namenodes [0.0.0.0]
0.0.0.0: starting secondarynamenode, logging to /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-secondarynamenode-debianvm.out

    On peut se connecter à l’interface web du NameNode à l’adresse:

    http://localhost:50070/

    On devrait voir une interface du type:

    Interface cluster Hadoop

    Erreur “Connection refused”

    Une erreur “Connection refused” peut se produire quand on essaie de démarrer Hadoop avec start-dfs.sh. Cette erreur est assez déroutante puisqu’elle peut se produire sans raison particulière alors que le cluster a déjà fonctionné.

    On peut s’apercevoir de cette erreur après exécution de start-dfs.sh et quand on essaie d’accéder à l’interface d’Hadoop en se connectant à l’adresse http://localhost:50070.

    Pour savoir s’il s’agit bien de la même erreur, il suffit d’aller dans le répertoire des fichiers de logs:

    vi /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-datanode-<nom de la machine>.log

    Hadoop refuse de démarrer et loggue une erreur du type:

    2017-09-16 09:41:21,836 WARN org.apache.hadoop.ipc.Client: Failed to connect to server: localhost/127.0.0.1:9000: retries get failed due to exceeded maximum allowed retries number: 10
java.net.ConnectException: Connection refused
	at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
	at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
	at org.apache.hadoop.net.SocketIOWithTimeout.connect(SocketIOWithTimeout.java:206)
	at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:531)
	at org.apache.hadoop.net.NetUtils.connect(NetUtils.java:495)
	at org.apache.hadoop.ipc.Client$Connection.setupConnection(Client.java:681)
	at org.apache.hadoop.ipc.Client$Connection.setupIOstreams(Client.java:777)
	at org.apache.hadoop.ipc.Client$Connection.access$3500(Client.java:409)
	at org.apache.hadoop.ipc.Client.getConnection(Client.java:1542)
	at org.apache.hadoop.ipc.Client.call(Client.java:1373)
	at org.apache.hadoop.ipc.Client.call(Client.java:1337)
	at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:227)
	at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:116)
	at com.sun.proxy.$Proxy15.versionRequest(Unknown Source)
	at org.apache.hadoop.hdfs.protocolPB.DatanodeProtocolClientSideTranslatorPB.versionRequest(DatanodeProtocolClientSideTranslatorPB.java:274)
	at org.apache.hadoop.hdfs.server.datanode.BPServiceActor.retrieveNamespaceInfo(BPServiceActor.java:215)
	at org.apache.hadoop.hdfs.server.datanode.BPServiceActor.connectToNNAndHandshake(BPServiceActor.java:261)
	at org.apache.hadoop.hdfs.server.datanode.BPServiceActor.run(BPServiceActor.java:746)
	at java.lang.Thread.run(Thread.java:748)

    Toutes les tentatives futures pour se connecter à l’adresse localhost/127.0.0.1:9000 ne fonctionnent pas:

    017-09-16 09:43:22,011 WARN org.apache.hadoop.hdfs.server.datanode.DataNode: Problem connecting to server: localhost/127.0.0.1:9000
2017-09-16 09:43:28,013 INFO org.apache.hadoop.ipc.Client: Retrying connect to server: localhost/127.0.0.1:9000. Already tried 0 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)
2017-09-16 09:43:29,013 INFO org.apache.hadoop.ipc.Client: Retrying connect to server: localhost/127.0.0.1:9000. Already tried 1 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)
etc...

    L’erreur semble provenir du fait qu’Hadoop est sensible aux redirections DNS effectuées par Linux entre l’adresse localhost dirigée vers 127.0.0.1 et le nom réseau de la machine dirigée vers 127.0.1.1.

    Tout d’abord avant de commencer à changer la configuration, il faut stopper l’exécution d’Hadoop et de YARN si ce n’est pas déjà fait (en tant qu’utilisateur “hduser”) en exécutant:

    /usr/hadoop/hadoop-2.8.1/sbin/stop-yarn.sh
/usr/hadoop/hadoop-2.8.1/sbin/stop-dfs.sh

    Ensuite, il faut connaître le nom réseau utilisé par Hadoop en exécutant:

    hostname --fqdn

    Le résultat est le nom réseau utilisé.

    Pour corriger le problème, il faut éditer le fichier /etc/hosts en tant qu’utilisateur root:

    su
vi /etc/hosts

    Par défaut le fichier se présente de cette façon:

    127.0.0.1	localhost
127.0.1.1	<nom réseau de la machine>

# The following lines are desirable for IPv6 capable hosts
::1     localhost ip6-localhost ip6-loopback
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters

    Il faut paramétrer la même adresse pour le nom réseau de la machine et commenter les lignes correspondant à IPv6:

    127.0.0.1	localhost
127.0.0.1	<nom réseau de la machine>

# The following lines are desirable for IPv6 capable hosts
#::1     localhost ip6-localhost ip6-loopback
#ff02::1 ip6-allnodes
#ff02::2 ip6-allrouters

    Il faut ensuite affiner la configuration pour que certains fichiers temporaires soient créés dans le répertoire de l’utilisateur “hduser”. Il faut éditer le fichier core-site.xml et ajouter quelques éléments de configuration:

    vi /usr/hadoop/hadoop-2.8.1/etc/hadoop/core-site.xml

    Après avoir ajouté les éléments, le fichier doit se présenter de cette façon:

    <configuration>
 <property>
   <name>fs.defaultFS</name>
   <value>hdfs://localhost:9000</value>
   <description>The name of the default file system.</description>
 </property>
 <property>
   <name>hadoop.tmp.dir</name>
   <value>/home/hduser/hadoop_data/hd-data/tmp</value>
 </property>
 <property>
   <name>fs.checkpoint.dir</name>
   <value>/home/hduser/hadoop_data/hd-data/snn</value>
 </property>
 <property>
   <name>dfs.data.dir</name>
   <value>/home/hduser/hadoop_data/hd-data/dn</value>
 </property>;
</configuration>

    Les parties en gras sont les parties nouvelles par rapport à la configuration précédente.

    Il faut ensuite créer le répertoire hadoop_data dans le répertoire de l’utilisateur “hduser” (cette étape doit être effectuée en tant qu’utilisateur “hduser” pour qu’Hadoop puisse écrire dans le répertoire):

    mkdir /home/hduser/hadoop_data

    Il faut ensuite supprimer les fichiers temporaires d’Hadoop (en tant qu’utilisateur root):

    rm -r /tmp/hadoop*

    Enfin, il faut reformater HDFS en exécutant (en tant qu’utilisateur “hduser”):

    /usr/hadoop/hadoop-2.8.1/hdfs namenode -format

    On peut ensuite essayer de démarrer Hadoop avec:

    /usr/hadoop/hadoop-2.8.1/sbin/start-dfs.sh

    Le fichier log du NameNode devrait ne plus comporter l’erreur “Connection refused”. On peut consulter ce fichier de log et les autres fichiers pour le vérifier. Ces fichiers se trouvent dans:

    /usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-datanode-<nom de la machine>.log
/usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-namenode-<nom de la machine>.log
/usr/hadoop/hadoop-2.8.1/logs/hadoop-hduser-secondarynamenode-<nom de la machine>.log

    On peut tenter de se connecter à l’interface web d’Hadoop sur:

    http://localhost:50070

    5. YARN

    Configuration de YARN

    Cette étape permet de configurer le gestionnaire de cluster YARN (pour Yet Another Resource Negotiator) en nœud simple (ie. single-node). Il faut éditer le fichier /usr/hadoop/hadoop-2.8.1/etc/hadoop/mapred-site.xml (ou le créer s’il n’existe pas en copiant mapred-site.xml.template) et ajouter:

    <configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>
</configuration>

    Cette configuration permet d’indiquer d’utiliser YARN en tant qu’implémentation de MapReduce.

    Il faut ensuite éditer le fichier /usr/hadoop/hadoop-2.8.1/etc/hadoop/yarn-site.xml et ajouter:

    <configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>
    </property>
</configuration>

    Exécution de YARN

    Après la configuration, on peut démarrer le daemon ResourceManager et NodeManager en tapant:

    start-yarn.sh

    Cette commande est dans un répertoire rajouté dans la variable d’environnement PATH donc elle doit être accessible de n’importe où. Ce fichier se trouve dans /usr/hadoop/hadoop-2.8.1/sbin/start-yarn.sh.

    L’interface web du ResourceManager est accessible à l’adresse:

    http://localhost:8088/

    L’interface est du type:

    Interface YARN

    6. Execution d’un job MapReduce

    Le job le plus couramment exécuté pour illustrer l’exécution de jobs MapReduce est le Wordcount (i.e. comptage de mots). Il faut utiliser en entrée un texte et en sortie du job, on obtient une liste de mots avec le nombre d’occurences pour chaque mot.

    Comme texte d’entrée, on peut utiliser le fichier correspondant au livre Guerre et Paix:
    http://www.gutenberg.org/files/2600/2600-0.txt

    Il faut copier ce fichier dans HDFS en écrivant:

    hdfs dfs -mkdir /user
hdfs dfs -mkdir /user/hduser
hdfs dfs -mkdir /user/hduser/input
hdfs dfs -put 2600-0.txt /user/hduser/input

    A ce moment on peut voir le fichier dans HDFS si on va dans l’interface web de Hadoop à l’adresse http://localhost:50070 dans l’onglet Utilities ➔ Browse the file system:

    HDFS

    On peut lancer le job en écrivant:

    hadoop jar /usr/hadoop/hadoop-2.8.1/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.8.1.jar wordcount /user/hduser/input/2600-0.txt output

    Cette ligne permet de lancer le job “wordcount” qui est implémenté dans le fichier JAR hadoop-examples-2.8.1.jar avec en entrée le fichier 2600-0.txt qui a été copié dans HDFS. Le résultat sera écrit dans le répertoire output dans HDFS.

    L’exécution donnera en sortie:

    17/09/09 12:16:44 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
17/09/09 12:16:45 INFO input.FileInputFormat: Total input files to process : 1
17/09/09 12:16:46 INFO mapreduce.JobSubmitter: number of splits:1
17/09/09 12:16:47 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1504949218149_0001
17/09/09 12:16:48 INFO impl.YarnClientImpl: Submitted application application_1504949218149_0001
17/09/09 12:16:48 INFO mapreduce.Job: The url to track the job: http://debianvm:8088/proxy/application_1504949218149_0001/
17/09/09 12:16:48 INFO mapreduce.Job: Running job: job_1504949218149_0001
17/09/09 12:17:09 INFO mapreduce.Job: Job job_1504949218149_0001 running in uber mode : false
17/09/09 12:17:09 INFO mapreduce.Job:  map 0% reduce 0%
17/09/09 12:17:39 INFO mapreduce.Job:  map 100% reduce 0%
17/09/09 12:17:53 INFO mapreduce.Job:  map 100% reduce 100%
17/09/09 12:17:57 INFO mapreduce.Job: Job job_1504949218149_0001 completed successfully
17/09/09 12:17:57 INFO mapreduce.Job: Counters: 49
	File System Counters
		FILE: Number of bytes read=649694
		FILE: Number of bytes written=1571945
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=3359665
		HDFS: Number of bytes written=487290
		HDFS: Number of read operations=6
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=2
	Job Counters 
		Launched map tasks=1
		Launched reduce tasks=1
		Data-local map tasks=1
		Total time spent by all maps in occupied slots (ms)=27460
		Total time spent by all reduces in occupied slots (ms)=10545
		Total time spent by all map tasks (ms)=27460
		Total time spent by all reduce tasks (ms)=10545
		Total vcore-milliseconds taken by all map tasks=27460
		Total vcore-milliseconds taken by all reduce tasks=10545
		Total megabyte-milliseconds taken by all map tasks=28119040
		Total megabyte-milliseconds taken by all reduce tasks=10798080
	Map-Reduce Framework
		Map input records=66055
		Map output records=566308
		Map output bytes=5541955
		Map output materialized bytes=649694
		Input split bytes=115
		Combine input records=566308
		Combine output records=41991
		Reduce input groups=41991
		Reduce shuffle bytes=649694
		Reduce input records=41991
		Reduce output records=41991
		Spilled Records=83982
		Shuffled Maps =1
		Failed Shuffles=0
		Merged Map outputs=1
		GC time elapsed (ms)=210
		CPU time spent (ms)=6020
		Physical memory (bytes) snapshot=412577792
		Virtual memory (bytes) snapshot=3924443136
		Total committed heap usage (bytes)=284688384
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=3359550
	File Output Format Counters 
		Bytes Written=487290

    On peut voir le fichier contenant les résultats en écrivant:

    hdfs dfs -ls /user/hduser/output

    Les fichiers sur HDFS sont:

    Found 2 items
-rw-r--r--   1 hduser supergroup          0 2017-09-09 12:17 /user/hduser/output/_SUCCESS
-rw-r--r--   1 hduser supergroup     487290 2017-09-09 12:17 /user/hduser/output/part-r-00000

    On peut récupérer le fichier contenant les résultats en écrivant:

    hdfs dfs -get /user/hduser/output/part-r-00000

    On peut lire maintenant le fichier. Le contenu est une liste de mots avec le nombre d’occurences pour chaque mot.

    Pour supprimer les fichiers se trouvant sur HDFS, il faut exécuter:

    hdfs dfs -rm -R /user/hduser/output
hdfs dfs -rm -R /user/hduser/input

    7. Stopper l’exécution de YARN et Hadoop

    Il faut taper les commandes suivantes:

    stop-yarn.sh

    Si l’arrêt se passe bien, l’exécution est du type:

    stopping yarn daemons
stopping resourcemanager
localhost: stopping nodemanager
localhost: nodemanager did not stop gracefully after 5 seconds: killing with kill -9
no proxyserver to stop

    stop-dfs.sh

    Le résultat peut être du type:

    Stopping namenodes on [localhost]
localhost: stopping namenode
localhost: stopping datanode
Stopping secondary namenodes [0.0.0.0]
0.0.0.0: stopping secondarynamenode
    Références

    Installation de Debian sur une machine virtuelle VirtualBox

    Le but de cet article est d’expliquer l’installation de Debian GNU/Linux 9 sur une machine virtuelle avec VirtualBox. Utiliser une machine virtuelle peut avoir de nombreux avantages quand on cherche à tester de nouvelles technologies:

    • On peut exporter la machine virtuelle si on souhaite partager une configuration.
    • En cas de mauvaise installation, on peut relancer une nouvelle instance d’une machine virtuelle et recommencer l’installation.
    • On ne pollue pas son système d’exploitation principal avec des installations qui ne vont pas forcément servir par la suite. On peut être libre de tester de nouveaux outils sans craindre qu’ils soient compliqués à désinstaller.

    GNU Debian/Linux et VirtualBox sont gratuits et peuvent être utilisés sans problématique de licence. Cette configuration permet de tester de nombreux outils qui sont installables dans un environnement Linux comme Docker, .NET Core, Teamcity, Hadoop, Spark, MongoDB, etc…

    Enfin, VirtualBox peut être installé sur tous les environnements ce qui donne davantage de flexibilité.

    Sommaire

    1. Installation de VirtualBox
    2. Créer une nouvelle VM
    3. Ajuster quelques paramètres sur la VM
    4. Télécharger Debian 9
    5. Installer Debian
    6. Démarrage de Debian
    7. Installation des “VirtualBox Guest Additions”

    1. Installation de VirtualBox

    La prmeière étape consiste à installer Oracle VM VirtualBox. Cet outil peut être téléchargé sur:
    https://www.virtualbox.org/
    Quelque soit la plateforme, cette étape ne présente pas de difficulté particulière.

    2. Créer une nouvelle VM

    Pour créer une machine virtuelle (i.e. VM) dans VirtualBox, il faut:

    • Cliquer sur “Nouvelle” puis choisir le type “Linux” et version “Debian (64-bit)”
    • Taille de la mémoire: 1024 Mo au minimum (ce paramètre peut être ajusté par la suite) toutefois on peut paramétrer une valeur plus confortable (par exemple 4096 Mo).
    • Pour la taille du disque, il faut prévoir une assez grande taille dès le début pour éviter d’être trop à l’étroit. Ce paramètre ne pourra pas être ajusté par l’interface graphique pas la suite. Il est possible d’augmenter la taille du disque après en utilisant l’instruction: 
      VBoxManage modifyhd --resizebyte <taille voulue en MO> <chemin du fichier VDI>

      Toutefois cette ligne ne fonctionne pas toujours.

    Création de la machine virtuelle
    Ajustement du paramètre de mémoire

    Pour le disque dur donc choisir: “Créer un disque dur virtuel maintenant”.
    Puis “VDI (Image Disque VirtualBox)”.
    Et “Dynamiquement alloué”.
    Ensuite ajuster la taille à 30 GO pour ne pas manquer d’espace disque (cette taille n’est pas allouée tout de suite, elle ne sera utilisée que si la VM consomme réellement l’espace).

    Création du disque dur

    3. Ajuster quelques paramètres sur la VM

    On peut affiner quelques paramètres dans la VM en effectuant un clique droit sur la VM puis en cliquant sur “Configuration”.

    Par exemple, on peut affiner le réglage de la mémoire vidéo dans l’onglet “Affichage” puis “Ecran” (ce paramétrage peut être ajusté après la création de la VM).

    Ajustement de la mémoire vidéo

    On peut aussi ajuster le nombre de coeurs du processeur dans l’onglet “Système” puis “Processeur”:

    • Sélectionnez “4”.
    • Valider en cliquant sur “OK”.
    Réglage du nombre de processeur

    4. Télécharger Debian 9

    Il faut ensuite télécharger l’image ISO d’installation de GNU Debian/Linux sur https://www.debian.org/:

    1. Aller dans “Images ISO cd CD/USB”.
    2. Choisir la méthode de téléchargement, par exemple “Télécharger les images des CD ou DVD par HTTP ou FTP”
    3. Choisir la distribution “stable”, actuellement il s’agit de “Stretch” (Debian 9).
    4. Cliquer sur “images officielle des CD ou DVD de la distribution ‘stable’
    5. Choisir l’architecture, pour une machine 64 bits, ça sera “amd64”.
      L’image de base suffit, par exemple: “debian-9.x.x-amd64-netinst”

    5. Installer Debian

    Pour commencer l’installation de Debian sur la machine virtuelle, il faut effectuer les étapes suivantes:

    1. Effectuer un clique droit sur la VM puis cliquer sur “Configuration”.
    2. Aller dans l’onglet “Stockage” puis cliquer sur “Controleur IDE” puis “Vide”:
      Montage de l’image de Debian
    3. Cocher ensuite “Live CD/DVD” puis cliquer sur l’icone du disque et cliquer sur “Choisissez un fichier de disque optique virtuel…”:

      Sélectionner l’image de Debian
    4. Sélectionnez l’image ISO téléchargée puis valider par “OK”.
    5. Cliquer ensuite sur “OK” pour valider la configuration de la VM.
    6. Cliquer ensuite sur “Démarrer” pour lancer la VM:
      Démarrer la VM

    A ce stade l’installateur de Debian est lancé.
    Choisir “Graphical install”:

    Installateur de Debian

    L’installation est ensuite classique:

    1. Choisir la langue, par exemple “English”.
    2. “Select a location”: la position dans l’écran “Location”: cliquer sur “other” ➜ “Europe” ➜ “France”.
    3. “Configure locales”: par exemple “United Kingdom”.
    4. “Configure the keyboard”: sélectionnez “French”.
    5. “Configure the network”: indiquer un nom réseau, par exemple “debianvm”.
      indiquer ensuite un nom de domaine: ce paramètre peut rester vide:

      Ecran “Configure the network”
    6. “Set up users and passwords”: choisir un mot de passe pour l’utilisateur root.
      choisir ensuite le nom d’un utilisateur, son nom de compte puis son mot de passe.

      Ecran “Set up users and passwords”
    7. “Partition disks”: choisir “Guided partitioning” ➜ “Guided – use entire disk” ➜ Sélectionnez ensuite le disque présent.
      Ecran “Partition disks (1/6)”
      Ecran “Partition disks (2/6)”
      Ecran “Partition disks (3/6)”
    8. Sélectionez ensuite “All files in one partition (recommended for new users)”.
      Puis “Finish partitioning and write changes to disk”.

      Ecran “Partition disks (4/6)”
      Ecran “Partition disks (5/6)”
    9. A la question “Write the changes to disks”, sélectionnez “Yes”:
      Ecran “Partition disks (6/6)”
    10. “Configure the package manager”: sélectionnez “No” puis “France” et “ftp.fr.debian.org”.
      Ecran “Configure the package Manager (1/4)”
      Ecran “Configure the package Manager (2/4)”
      Ecran “Configure the package Manager (3/4)”
    11. Indiquez éventuellement des informations de proxy:
      Ecran “Configure the package Manager (4/4)”
    12. “Configuring popularity-contest”: sélectionnez la valeur de votre choix.
    13. “Software selection”:
      Cochez “Debian desktop environment” et “standard system utilities”
      Décochez “print server”

      GNOME sélectionné par défaut

      Avec ce paramétrage, GNOME sera sélectionné par défaut. Bien que GNOME soit un environnement de bureaux facile à utiliser, il fait partie des environnements les plus lourds. Pour économiser les ressources on peut, par exemple, sélectionner Xfce qui est plus léger mais moins riche en utilitaires.

      Ecran “Software selection”
    14. “Install the GRUB boot loader on a hard disk”:
      Cliquer sur “Yes” pour installer GRUB:

      Ecran “Install the GRUB boot loader on a hard disk (1/2)”
    15. Sélectionnez le disque c’est-à-dire “/dev/sda”:
      Ecran “Install the GRUB boot loader on a hard disk (2/2)”

    L’installation se termine par:

    Fin de l’installation

    6. Démarrage de Debian

    A la fin de l’installation, Debian redémarre automatiquement.
    Si ce n’est pas le cas, il faut démarrer manuellement la VM mais il faut enlever le disque virtuel d’installation en allant dans la configuration de la VM:

    1. Faire un clique droit sur la VM puis “Configuration”.
    2. Allez dans l’onglet “Stockage” puis sélectionnez le disque ISO “debian-9.x.x-amd64-netinst.iso”.
      Décochez “Live CV/DVD”
      Cliquez sur l’icone du disque puis clique sur “Retirer le disque du lecteur virtuel”
      Valider en cliquant sur “OK”.

      Retirer le disque Debian
    3. Démarrer ensuite la VM en cliquant sur Démarrer.

    On arrive ensuite sur l’invite GRUB de sélection du système à démarrer. La sélection est automatique, il suffit d’attendre quelques secondes:

    Invite GRUB

    Entrez le mot de passe si nécessaire en cliquant sur le nom de l’utilisateur:

    Mot de passe pour l’ouverture de session

    On arrive ensuite sur l’interface de Debian:

    Interface de GNOME

    7. Installation des “VirtualBox Guest Additions”

    Les VirtualBox Guest Additions sont des composants qui permettent de personnaliser des éléments de configuration dans Debian en accord avec la VirtualBox. Par exemple, pour configurer correctement la résolution de la VM.

    Il faut télécharger l’image ISO des VirtualBox Guest Additions sur:
    http://download.virtualbox.org/virtualbox/

    Il faut cliquer ensuite sur la version de la VirtualBox puis sélectionnez “VBoxGuestAdditions_5.x.x.iso”.

    Dans la VM Debian, il faut ouvrir un terminal:

    1. Cliquer sur Activities en haut à gauche puis écrire terminal et cliquer sur l’icone du terminal:
      Ouverture d’un terminal (1/2)
      Ouverture d’un terminal (2/2)
    2. Taper ensuite su pour se connecter comme utilisateur root.
    3. Entrez ensuite le mot de passe de l’utilisateur root.
    4. Taper apt-get update pour mettre à jour la liste des packages du gestionnaire de packages Debian APT.

    Il faut installer ensuite les packages “build-essential” et “module-assistant”.
    Ces packages sont nécessaires pour être capable de compiler les modules du noyau qui sont nécessaires à l’installation de VirtualBox Linux Guest Additions.

    Pour installer ces packages, il faut taper:

    apt-get install build-essential module-assistant

    A cette étape, une question est posée car la quantité de packages à installer est importante.
    Il faut répondre en indiquant [Y] puis [entrée]:

    Installation des packages “build-essential” et “module-assistant”

    Quand l’installation est terminée, il faut configurer le système pour construire les modules du noyau en tapant:

    m-a prepare

    Lors de cette étape, de nouveaux packages doivent être installés comme les fichers headers du noyau linux.
    Il faut répondre en tapant [Y] puis [Entrée].

    Execution de l’instruction m-a prepare

    Il faut ensuite insérer le disque virtuel de l’image ISO des VirtualBox Guest Additions en allant dans les paramètres de la VM dans VirtualBox:

    1. Clique droit sur la VM puis cliquez sur “Configuration”
    2. Dans l’onglet “Stockage”, sélectionnez “Vide” sous “Controleur: IDE”
    3. Cliquez sur l’icone du disque à droite de la fenêtre puis cliquez sur: “Choisissez un fichier de disque optique virtuel…”
      Sélectionner le disque VirtualBox Guest Additions (1/2)
      Sélectionner le disque VirtualBox Guest Additions (2/2)
    4. Sélectionnez le fichier ISO “VBoxGuestAdditions-5.x.x.iso” puis validez en cliquant sur “OK”.

    A cette étape, le disque devrait être monté dans la VM. Si ce n’est pas le cas, on peut taper la commande suivante pour le monter:

    mount /media/cdrom

    Il faut ensuite exécuter un script pour compiler les modules additionnels correspondant au noyau en tapant:

    sh /media/cdrom/VBoxLinuxAdditions.run
    Exécution du script de compilation des modules additionnels

    On redémarre la VM en tapant:

    reboot

    On peut maintenant facilement paramétrer la résolution en cliquant sur “Activities” en haut à gauche et en tapant settings:

    Accéder au panneau “Settings”
    1. Cliquer ensuite “Settings” puis “Displays”.
    2. Sélectionnez l’écran en cliquant dessus.
    3. Cliquer sur la résolution qui convient et valider en cliquant sur “Apply”.
      4. Sélection de la résolution

    On peut enlever le disque virtuel en tapant sur un terminal en tant qu’utilisateur root:

    eject /media/cdrom

    A la fin de cette étape, la VM est opérationnelle et utilisable en mode plein écran.

    Pour tirer partie de la flexibilité de la VM, on peut la cloner en effectuant un clique droit sur son icone et en cliquant ensuite sur “Cloner…”.
    On peut aussi exporter une VM en cliquant sur “Fichier” puis “Exporter un appareil virtuel…”.

    L’intérêt de cloner ou d’exporter une VM est de pouvoir repartir d’une version “vierge” lors des différents tests d’installation. Cette possibilité peut s’avérer très utile si on se trompe d’installation ou si on doit repartir de zéro.

    Il faut noter que si on supprime la VM dans VirtualBox, on ne pourra pas la réintégrer par le suite. Il est donc préférable de cloner ou d’exporter une VM pour la réutiliser par la suite.

    Références

    Powershell en 10 min: références

    PARTIE 1 PARTIE 2 PARTIE 3 PARTIE 4
    Sommaire général Exécuter Powershell Les cmdlets Instructions dans des scripts Aide, gestion d’erreurs et manipulation de fichiers Références

    Références

    Powershell en 10 min

    PARTIE 1 PARTIE 2 PARTIE 3 PARTIE 4
    Sommaire général Exécuter Powershell Les cmdlets Instructions dans des scripts Aide, gestion d’erreurs et manipulation de fichiers Références

    Le but de cet article est de présenter succinctement les fonctionnalités principales de Powershell de façon à pouvoir écrire des scripts et tirer rapidement partie de ce langage.

    Sommaire

    Partie 1: exécuter du code Powershell
    Installation de Powershell
    Utilisation de la console
    Exécuter un script Powershell
    Utiliser un environnement pour développer des scripts Powershell

    Partie 2: les “cmdlets”
    PowerShell Module Browser
    Alias
    Quelques exemples de “cmdlets”
    Chaines de caractères
    Utiliser le Framework .NET
    Utiliser des “pipelines”
    Powershell “providers” (utilisation des “snap-ins”)

    Partie 3: instructions dans des scripts Powershell
    Manipulation de variables
    if…then…else
    switch
    Définition des listes d’éléments
    Boucles
    Scripts blocks

    Partie 4: aide, gestion d’erreurs et manipulation de fichiers
    Messages d’aide
    Gestion des erreurs
    Manipulation de fichiers

    Références

    Pour commencer

    Partie 1: exécuter Powershell

    Partie 2: les cmdlets

    Partie 3: instructions dans des scripts Powershell

    Partie 4: aide, gestion d’erreurs et manipulation de fichiers

    Références