Ca y est, vous avez décidé de mettre en place un cluster Hadoop.

Prochaine étape, le dimensionnement... Hadoop étant une solution complexe, plusieurs questions se posent :

• HDFS gère des réplicas, Map Reduce génère des fichiers, comment faire pour prévoir mon stockage ?
• Comment prévoir mes besoins en CPU ?
• Comment prévoir mes besoins en mémoire ? Faut il faire une distinction sur certaines parties du cluster ?
• On m'a dit que Map Reduce déplace le code proche des fichiers... Concrètement, qu'est ce que cela implique pour prévoir mes besoins réseau ?
• Dans quelle mesure les cas d'usages métier entre en compte dans le dimensionnement ?

C'est ce que nous allons tenter d'éclaircir dans cet article en fournissant des explications sur ces différents points ainsi que des moyens pour calculer vos besoins.

Quelques principes forts de l'architecture d'Hadoop

Dimensionner un cluster Hadoop demande de comprendre certains fondamentaux de son architecture.

Le calcul distribué

En son coeur, Hadoop est une plateforme de calcul distribué. Cette plateforme de calcul se base sur le modèle de programmation parallèle de MapReduce.

Pour être efficient, MapReduce a deux prérequis :

• Les jeux de données doivent pouvoir être découpés en petits blocs unitaires et indépendants les uns des autres
• Les traitements doivent pouvoir être déplacés là où se trouvent les données, pas le contraire

Le premier prérequis dépend du type de données dont on alimente le cluster et de ce que l'on veut en faire.

Le second prérequis nécessite de disposer à la fois d'un système de stockage disponible, vaste et permettant d'exécuter du code là où se trouve les blocs de données.

C'est à ce second prérequis que répond la brique HDFS.

L'architecture d'Hadoop est de type Maître / Esclave :

• Le JobTracker (_ResourceManager_dans Hadoop 2)
  • Gère les jobs s'exécutant sur le cluster
  • Nécessite beaucoup de mémoire et de CPU (memory bound et cpu bound)
• Le TaskTracker (NodeManager + _ApplicationMaster_dans Hadoop 2)
  • Exécute les tâches d'un job sur chaque noeud du cluster. Il s'agit des Map et des Reduces
  • Ses jobs nécessitent beaucoup de mémoire et de CPU (memory bound et cpu bound)

Le composant critique dans cette architecture est le JobTracker/ResourceManager.

Le stockage distribué

HDFS est un système de stockage distribué. Il s'appuie sur le système de fichiers du système d'exploitation.

Pour être efficient, HDFS a 3 prérequis :

• Des disques rapides en terme de débit sur des lectures et écritures séquentielles de gros blocs de données (64Mo, 128Mo ou encore 256Mo)
• Un système de fichiers supportant ce mode d'utilisation de manière intensive
• Un réseau performant pour transférer les jeux de données et gérer la réplication

L'architecture d'HDFS est de type Maître / Esclave :

• Le NameNode et le Secondary NameNode
  • Gèrent les méta informations (répertoires, noms, attributs et localisation des fichiers) ainsi que la réplication des blocs du cluster
  • Nécessite beaucoup de mémoire (memory bound)
• Le DataNode
  • Gère l'état d'un noeud HDFS et permet d'intéragir avec ses blocs
  • Nécéssite beaucoup d'I/O lors des traitements et des transferts de données (I/O bound)

Les composants critiques dans cette architecture sont le NameNode et le Secondary NameNode.

Nous avons donc deux architectures distinctes bien qu'intimement liées.

Il est d'ailleurs possible de ne pas utiliser HDFS avec Hadoop. Par exemple, Amazon, avec son offre Elastic MapReduce se base sur son offre de stockage S3 et un outil comme KarmaSphere Analyst embarque un moteur Hadoop se basant sur un répertoire local plutôt que sur HDFS.

Quelques détails techniques qui comptent pour un dimensionnement

Sans être un expert d'Hadoop, quelques points techniques sont à connaître lorsque l'on souhaite dimensionner son cluster.

Comment HDFS gère des fichiers

HDFS est optimisé pour le stockage de gros fichiers. On y écrit une fois le fichier, puis on y accède plusieurs fois.

Dans HDFS, un fichier est découpé en bloc. Chaque bloc est répliqué dans le cluster de manière asynchrone. L'utilisateur envoi ainsi une seule fois son fichier et le cluster se charge seul de répliquer les blocs qui doivent l'être.

Un bloc est un zone continue sur le système de fichiers sous jacent. La taille par défaut est de 64Mo mais elle peut être étendue à 128Mo voire 256Mo en fonction des besoins.

La réplication des blocs, qui est de 3 par défaut, a deux fonctions :

• Assurer que les données ne seront pas perdues lors de la perte d'un DataNode. A ce titre, les disques utilisés pour HDFS ne doivent pas être mis en configuration RAID
• Augmenter le nombre de map traitant un bloc lors d'un job MapReduce et ainsi accélerer le traitement

D'un point de vu réseau, la bande passante sera donc sollicitée aux moments :

• De la réplication suivant un dépôt de fichier
• De l'équilibrage du facteur de réplication en cas de perte d'un DataNode

Comment le NameNode gère le cluster HDFS

Le NameNode gère les méta informations du cluster HDFS.

Cela inclu les méta informations (noms des fichiers, des répertoires, ...) ainsi que la localisation des blocs composants un fichier.

Cette gestion est effectuée en mémoire et un fichier fsimage ainsi qu'un fichier edits contenant la liste des modifications est écrit sur le disque dur pour lui permettre de reconstruire l'état du cluster en mémoire en cas de redémarrage.

Le rôle du Secondary NameNode est de mettre à jour le fichier fsimage à interval régulier en y intégrant le contenu des edits.

Le déclencheur pouvant être :

• le nombre de transactions effectuées sur le cluster
• la taille du fichier d'edits
• le temps écoulé depuis la dernière compaction

Pour déterminer le besoin en mémoire d'un NameNode pour gérer le cluster HDFS, on applique la formule suivante :

<RAM pour gérer le cluster HDFS> = <taille totale du stockage du cluster en Mo> / <taille d'un bloc en Mo> / 1000000

En d'autres termes, on considère que le NameNode a besoin d'un Go par million de bloc géré.

Comment calculer les besoins en dimensionnement ?

Calculer ses besoins en stockage

Les besoins en stockage peuvent se découper en trois partie :

• Le tronc commun
• Les besoins des NameNode et Secondary NameNode
• Les besoins des DataNode

Le tronc commun

Le tronc commun est une partie déjà largement maîtrisée qui comprend :

• La partition d'installation de l'OS
• La partition des logs

Ces deux partitions peuvent être gérées comme pour n'importe quel autre projet dans votre DSI.

Les besoins des NameNode et Secondary NameNode

Le Secondary NameNode doit être à l'identique du NameNode.

Une partition d'1To dédiée aux fichiers écrits spécifiquement par le NameNode et le Secondary NameNode permet de faire croitre le cluster sans risque de saturation du disque.

Si vous souhaitez coller de plus près à la taille réelle occupée, il faut prendre en compte les paramétrages du NameNode décrit plus haut (quel déclencheur pour la compaction du fichier edits) et multiplier le résultat par le nombre de sauvegardes d'images compactées que vous souhaitez conserver.

Dans tous les cas, le NameNode doit avoir un point de montage distant sur un stockage sécurisé dans lequel il peut sauvegarder la dernière version de ses fichiers fsimage et edits.

Ce point de montage ayant la même taille que la partition dédiée mentionnée plus haut.

Le stockage des NameNode et Secondary NameNode est typiquement effectué sur des configuration en RAID.

Les besoins des DataNode

Le stockage des _DataNode_s doit être :

• Contrôleur SAS 6Gb/s en configuration JBOD (Just a Bunch of Disk)
• Disques SATA II de 1 à 3To à 7200 tpm

Le RAID n'est pas utilisé sur les DataNode, HDFS ayant son propre mécanisme de réplication.

Le nombre de disques varie en fonction de la capacité de stockage souhaitée.

Pour estimer cette dernière, une bonne manière de faire est de partir de l'alimentation prévisionnelle du cluster.

Il faut aussi prendre en compte pour chaque disque, 30% de besoin de stockage en plus pour la partie non HDFS.

Prenons les hypothèses suivantes :

• Alimentation quotidienne : 100Go
• Facteur de réplication HDFS : 3
• Espace réservé par disque hors HDFS : 30%
• Taille d'un disque : 3To

Avec ces hypothèses, nous sommes en mesure de prévoir l'espace de stockage nécessaire ainsi que le nombre de DataNodes.

Ainsi nous avons :

• Espace occupé sur le cluster par l'alimentation quotidienne : <alimentation quotidienne> * <facteur de réplication> = 300Go
• Taille d'un disque dédiée à HDFS : <taille d'un disque> * (1 - <espace réservé par disque hors HDFS>) = 2,1To
• Nombre de DataNodes à 1 an (sans croissance mensuelle) : <espace occupé sur le cluster par l'alimentation quotidienne> * 365 / <taille d'un disque dédiée à HDFS> = 100To / 2,1To = 48 DataNodes

Deux variables importantes ne sont pas inclues dans ce petit calcul :

• le taux de croissance mensuel de l'alimentation des données
• Le ratio de données générée par les jobs pour une alimentation

Ces informations dépendent des problématiques métiers et doivent être prises en compte dans une évaluation du besoin en stockage.

Calculer ses besoins en CPU

Sur les NameNode et Secondary NameNode, 4 coeurs physiques à 2Ghz suffiront amplement.

En ce qui concerne les DataNodes, deux éléments vous aident à déterminer vos besoins en CPU :

• Le type de travaux que le DataNode va effectuer
• Le nombre de jobs que vous souhaitez par DataNode

Le type de travaux

En simplifiant, on considère qu'un DataNode peut effectuer deux type de travaux : intensif en I/O et intensif en CPU.

Les travaux intensifs en I/O

Ces travaux consomment principalement de l'I/O. Il sont dit I/O bound.

Il s'agit par exemple :

• d'indexation
• de recherche
• de regroupement
• de décodage / décompression
• d'import / export de données

Pour ce type de travaux, un CPU entre 2Ghz et 2,4Ghz suffit.

Les travaux intensifs en CPU

Ces travaux consomment principalement du CPU. Il sont dit CPU bound.

Il s'agit par exemple :

• de machine learning
• de statistique
• d'analyse sémantique
• d'analyse linguistique

Pour ce type de travaux, un CPU entre 2,6Ghz et 3Ghz suffit.

Le nombre de jobs

Le nombre de coeurs physique détermine le nombre maximum de jobs que l'on peut faire tourner en parallèle sur un DataNode. Cela en gardant à l'esprit une répartition de ces tâches entre les tâches de Map et celles de Reduce (prenez 2/3 Map et 1/3 Reduce).

Pour calculer cela, on peut utiliser la formule suivante qui prend en compte :

(<nombre de coeurs physique> - 1) * 1,5 = <nombre maximum de tâches>

ou, si vous préférez partir du nombre de tâches souhaité pour ajuster le nombre de coeurs physique :

(<nombre maximum de tâches> / 1,5) + 1 = <nombre de coeurs physique>

Le 2 représente les processes TaskTracker (MapReduce) et DataNode (HDFS). Le 1,5 permet d'intégrer le fait que les CPUs physiques ont plusieurs threads (Hyperthreading).

Calculer ses besoins en mémoire

Il y a deux types de calculs a effectuer pour calculer ses besoins en mémoire :

• La mémoire des NameNode et Secondary NameNode
• La mémoire des DataNode

Dans tous les cas, privilégiez de la mémoire DDR3 ECC.

La mémoire des NameNode et Secondary NameNode

Comme expliqué plus haut, le NameNode gère le cluster HDFS en mémoire. A cela s'ajoute la mémoire pour le process NameNode ainsi que la mémoire à réserver pour le système d'exploitation.

Concernant le Secondary NameNode, il doit être à l'identique du NameNode.

Par conséquent, nous avons :

<RAM du Secondary NameNode> =  <RAM du NameNode> = <RAM pour gérer le cluster HDFS> + <2Go pour le process NameNode> + <4Go pour l'OS>

La mémoire des DataNode

La mémoire des DataNode se calcule en fonction du type de traitement prévu sur le cluster.

Sur un cluster I/O bound, un ratio de 2Go à 4Go par coeur physique.

Sur un cluster CPU bound, un ratio de 6Go à 8Go par coeur physique.

A cela s'ajoute :

• 2Go pour le process DataNode qui gère les blocs HDFS
• 2Go pour le process TaskTracker qui gère les tâches des jobs sur le noeud
• 4Go pour l'OS

Ce qui nous donne au final les deux formules suivantes :

<RAM du DataNode I/O bound> = 4Go * <nombre de coeurs physique> + <2Go pour le process DataNode> + <2Go pour le process TaskTracker> + <4Go pour l'OS>

<RAM du DataNode CPU bound> = 8Go * <nombre de coeurs physique> + <2Go pour le process DataNode> + <2Go pour le process TaskTracker> + <4Go pour l'OS>

Calculer ses besoins en réseau

Au niveau réseau, il y a deux sources principales de traffic dans Hadoop :

• La phase de shuffle pendant laquelle les données traitées par les mappers sont passées aux reducers
• L'équilibrage du facteur de réplication (suite à un ajout de fichier ou à la chute d'un DataNode)

De plus, les transferts réseau dans Hadoop se font beaucoup de manière transverse entre les DataNodes (East/West).

Tant que l'on reste intra rack, cela ne pose pas de soucis et Hadoop fait le maximum pour se limiter à ce type de transfert.

Cependant, des transferts inter racks ont également lieu par exemple pour le second réplica d'un bloc HDFS.

Le sujet est assez complexe, mais d'une manière générale, il est préférable de :

• Opter pour une architecture réseau de type Spine Fabric
  • Maximisation de la bande passante
  • Meilleure résistance aux pannes
• Eviter la sursouscription des switchs
  • Un rack avec 48 ports 1Gb doit avoir 5 ports 10Gb sur les switchs de distribution
  • Permet d'éviter les ralentissements du cluster sous forte charge (jobs + alimentations)
• Si le cluster est I/O bound ou si vous prévoyez des alimentations régulières saturant des liens 1Gb et ne pouvant être effectuées hors heures de travail (la nuit par exemple) utilisez des connexions :
  • 10Gb Ethernet intra rack
  • N x 10Gb Ethernet inter rack
• Si le cluster est CPU bound utilisez des connexions :
  • 2 x 1Gb Ethernet intra rack
  • 10Gb Ethernet inter rack