🐘 Apache Hadoop – HDFS, YARN & MapReduce

Qu'est-ce qu'Apache Hadoop ?

Apache Hadoop n'est pas un seul produit, c'est un framework (cadre) open-source (Java) pour le stockage et le traitement (Calcul) de très grands ensembles de données (Big Data) de manière distribuée sur des clusters de matériel "banal" (commodity hardware).

Il est basé sur les papiers de recherche de Google (GFS et MapReduce) de 2003-2004.

Les 3 Piliers (Hadoop 2.0+)

L'écosystème Hadoop "moderne" (Hadoop 2.0+) repose sur 3 composants clés :

1. HDFS (Stockage) : (Voir 2.1) Hadoop Distributed File System. Le système de fichiers (Data Plane) qui stocke les données (fichiers) de manière distribuée et répliquée.
2. YARN (Gestion des Ressources) : (Voir 4.1) Yet Another Resource Negotiator. Le "cerveau" (Control Plane) qui gère les ressources (CPU, RAM) du cluster.
3. MapReduce (Modèle de Calcul) : (Voir 5.1) Le "moteur" de calcul (Compute) (legacy) qui traite les données (basé sur disque).

Hadoop (Big Data) a été conçu pour résoudre les problèmes que les bases de données relationnelles (SQL) ne pouvaient pas gérer.

Apache Spark (voir guide Spark) a été conçu pour remplacer le composant MapReduce (Compute) de Hadoop.

Hadoop n'est pas mort : Spark (Compute) s'exécute (très souvent) sur YARN (Resource Manager) et lit/écrit sur HDFS (Storage).

Le HDFS (Hadoop Distributed File System) est la couche Stockage de Hadoop. C'est un système de fichiers (comme NTFS ou EXT4) mais distribué (sur des milliers de serveurs).

Philosophie (WORM & Big)

HDFS est conçu (optimisé) pour :

• Fichiers Très volumineux : (Giga-octets à Péta-octets).
• WORM (Write-Once, Read-Many) : On écrit le fichier 1 fois (en entier), et on le lit N fois. (Les fichiers HDFS ne sont (généralement) pas modifiables (append-only)).
• Streaming (Lecture Séquentielle) : Optimisé pour des scans (SELECT *) massifs, pas pour des accès aléatoires (WHERE id=1).
• Matériel Banal (Commodity Hardware) : Il est conçu pour tolérer les pannes (disques, serveurs) (via la réplication).

HDFS utilise une architecture Maître/Esclave (Leader/Follower).

(Client : "Lire /data/file.txt")
       │
       │ 1. (Demande Métadonnées)
       ▼
[ NAMENODE (Maître) ]
 (Le "Cerveau")
 (Stocke Métadonnées : Arborescence,
  Permissions, Blocs...)
       │
       │ 2. (Réponse : "Le Fichier est sur
       │     DN1 (Bloc A) et DN3 (Bloc B)")
       ▼
(Client) ◄── 3. (Lit Bloc A) ── [ DATANODE 1 (Esclave) ] (Stocke Blocs)
   │
   └─────── 4. (Lit Bloc B) ──► [ DATANODE 3 (Esclave) ] (Stocke Blocs)

   [ DATANODE 2 (Esclave) ] (Stocke Blocs)

• NameNode (Maître) : (1 seul, ou 2 en HA). Le "cerveau". Ne stocke aucune donnée. Il stocke uniquement les Métadonnées (l'arborescence /, les permissions chmod, et la "carte" (quels blocs composent un fichier, et où sont ces blocs)). C'est le SPOF (Single Point of Failure) (voir 3.3).
• DataNode (Esclave) : (N nœuds) Les "muscles". Ils ne font que stocker les Blocs (128MB) (les "vrai" données) sur leurs disques locaux. Ils envoient des "Heartbeats" (signaux de vie) et des "Block Reports" (liste de leurs blocs) au NameNode.

Gros Blocs (128 Mo)

HDFS (comme ext4) divise les fichiers en Blocs. Mais les blocs HDFS sont énormes (par défaut 128 Mo ou 256 Mo).

Exemple (Fichier de 300 Mo) :

• Bloc A : 128 Mo
• Bloc B : 128 Mo
• Bloc C : 44 Mo

Réplication (Tolérance aux Pannes)

C'est la clé de la fiabilité d'HDFS (conçu pour du matériel "banal" qui tombe en panne).

Par défaut, le Facteur de Réplication est 3. HDFS copie chaque bloc 3 fois, sur 3 DataNodes différents (idéalement sur 3 Racks différents, voir Rack Awareness).

(Fichier 300MB -> Blocs A, B, C)

[ DataNode 1 ] -> (Bloc A), (Bloc C)
[ DataNode 2 ] -> (Bloc B)
[ DataNode 3 ] -> (Bloc A), (Bloc B)
[ DataNode 4 ] -> (Bloc C), (Bloc A)
[ DataNode 5 ] -> (Bloc B), (Bloc C)

Panne : Le DataNode 2 (panne disque) meurt. Le NameNode le détecte (pas de Heartbeat).

Auto-Réparation : Le NameNode voit que le Bloc B n'est plus répliqué que 2x (sur DN3, DN5). Il ordonne (automatiquement) à DN4 de copier le Bloc B depuis DN3. La réplication (3x) est restaurée.

Le NameNode (Maître) est le SPOF (Point de Défaillance Unique). Il stocke toutes ses métadonnées (l'arborescence) en RAM (pour la vitesse).

Si la RAM du NameNode est pleine, le cluster est plein (même s'il reste de l'espace sur les DataNodes).

Persistance (fsimage & EditLog)

Pour persister (sur disque) les métadonnées (qui sont en RAM), le NameNode utilise deux fichiers :

État Actuel (RAM) = fsimage (lu au boot) + EditLog (rejoué au boot).

Piège : Le Secondary NameNode (SNN) n'est PAS un "Backup" (Hot-Standby) du NameNode. Il ne prend pas le relais si le NameNode (NN) crash.

Rôle : Le "Checkpointing" (Fusion)

Problème : Si le cluster tourne 1 an, le fsimage (Snapshot) a 1 an, et le EditLog (Journal) pèse 50 Go. Si le NameNode redémarre, il doit "rejouer" 50 Go d'opérations (peut prendre 1 heure).

Solution (SNN) : Le SNN (qui tourne sur une autre machine) effectue le "checkpointing" (la fusion) :

1. (Toutes les heures) SNN contacte NN : "Crée un nouveau EditLog.new".
2. SNN télécharge l'ancien fsimage et l'ancien EditLog (plein).
3. SNN (localement) fusionne (applique) fsimage + EditLog pour créer un nouveau fsimage.ckpt.
4. SNN "upload" (envoie) le nouveau fsimage.ckpt au NameNode.
5. (Au prochain boot, le NN utilise ce fsimage récent et le petit EditLog.new -> Démarrage rapide).

Problème : Le NameNode (3.1) est un SPOF. Le Secondary NN (3.2) n'est pas un backup (failover).

Solution (Hadoop 2.0+) : NN (Active/Standby)

On exécute deux NameNodes (NN) :

[ NN 1 (ACTIVE) ] <--- (Heartbeat/Lock) ---> [ ZooKeeper ]
       │                                         ▲
       │ (Écrit 50%)                             │ (Heartbeat/Lock)
       ▼                                         │
[ Quorum Journal Manager (QJM) ] (Stockage 'EditLog' partagé)
 (JournalNodes 1, 2, 3)
       ▲
       │ (Écrit 50%)
       │
[ NN 2 (STANDBY) ]

• NameNode (Active) : Gère le cluster.
• NameNode (Standby) : "Hot spare". Lit le QJM pour être à jour.
• QJM (Quorum Journal Manager) : Un stockage (léger) partagé (ex: 3 "JournalNodes") où les deux NNs écrivent (et lisent) l'EditLog (le journal). (Requiert (N+1)/2 acquittements).
• ZooKeeper (ZKFC) : (Le "Lock") Un service de consensus (externe) qui gère le verrou (lock). (Le NN qui "possède" le verrou dans ZK est l'ACTIVE).

Flux de Panne

1. NN 1 (Active) crash.
2. Sa session (Heartbeat) avec ZooKeeper expire.
3. NN 2 (Standby) voit le "lock" ZK se libérer.
4. NN 2 (Standby) prend le "lock" ZK et passe en ACTIVE. (Failover en ~10-30 sec).

YARN (Yet Another Resource Negotiator) (introduit dans Hadoop 2.0) est la couche de gestion de ressources (Compute) de Hadoop.

Le Problème (Hadoop 1.0)

Dans Hadoop 1.0, le Stockage (HDFS) et le Calcul (MapReduce) étaient couplés dans le "JobTracker" (Maître) et le "TaskTracker" (Esclave).

Inconvénient : Le cluster ne pouvait exécuter que du MapReduce. (Pas de Spark, pas de Flink...).

La Solution (Hadoop 2.0 - YARN)

YARN sépare le "Resource Management" (Gestion des Ressources) du "Processing" (Traitement).

[ COUCHE 2 : COMPUTE (Moteurs) ]
  (Ex: MapReduce, Spark, Flink, Tez...)
   │
   │ (Demandent des ressources)
   ▼
[ COUCHE 1 : YARN (Resource Manager) ]
 (Gère CPU/RAM du cluster)
   │
   │ (Utilise le stockage)
   ▼
[ COUCHE 0 : HDFS (Stockage) ]

YARN est le "Scheduler" (Planificateur) du cluster. Spark (ou MapReduce) demande à YARN : "J'ai besoin de 50 'Containers' (CPU/RAM) pour ce job". YARN les lui fournit (sur les NodeManagers).

YARN (comme HDFS) utilise une architecture Maître/Esclave.

(Client : "Lancer App Spark")
       │
       │ 1. (Demande "ApplicationMaster")
       ▼
[ RESOURCEMANAGER (RM) (Maître) ]
 (Le "Cerveau" YARN)
 (Gère *toutes* les ressources (CPU/RAM))
       │
       │ 2. (Négocie avec un NM pour lancer AM)
       ▼
[ NODEMANAGER (NM) (Esclave) ] (Node 1)
   (Gère 1 Nœud (VM))
   (Contient des "Containers")
   │
   └─ [ 3. ApplicationMaster (AM) ] (1er Container)
           (Le "Cerveau" du Job)
           │
           │ 4. (Demande 10 Containers au RM)
           ▼
[ RESOURCEMANAGER (RM) ]
           │
           │ 5. (Alloue 10 Containers aux NMs)
           ▼
[ NODEMANAGER (NM) (Esclave) ] (Node 1...N)
   └─ [ 6. Container (Task) ] (Executor Spark)

• ResourceManager (RM) : (1 Maître) Le "cerveau" global. Il sait combien de CPU/RAM sont disponibles sur l'ensemble du cluster.
• NodeManager (NM) : (N Esclaves) Le "muscle" (tourne sur chaque DataNode). Il gère les ressources (CPU/RAM) de sa machine. Il lance/tue les "Containers" (processus) sur ordre du RM.

YARN (contrairement à Hadoop 1.0) introduit le concept d'"ApplicationMaster" (AM).

Container (Conteneur)

Ce n'est pas un "Conteneur Docker". C'est une réservation (slice) de ressources (CPU/RAM) sur un NodeManager. C'est là que le "vrai" code (ex: Tâche Map/Reduce, Executor Spark) s'exécute.

ApplicationMaster (AM)

L'AM est le "cerveau" spécifique à 1 Job (Application). C'est un "Driver" (Spark) ou "JobTracker" (MapReduce) dynamique.

Flux (YARN)

1. (Client) spark-submit.
2. (YARN) ResourceManager (RM) reçoit la demande.
3. RM demande à un NodeManager (NM) de lancer 1 Container pour l'ApplicationMaster (AM).
4. L'AM (le "cerveau" du Job) démarre.
5. L'AM négocie (demande) directement avec le RM : "J'ai besoin de 50 Containers (1vCPU/4Go RAM)".
6. Le RM alloue 50 Containers (sur 10 NMs).
7. L'AM contacte directement les NMs pour lancer les Tâches (Tasks) dans ces Containers.

Avantage : Le RM (Maître) ne fait que la "gestion" (RH). L'AM (Spécifique au Job) fait la "coordination" (Chef de projet). (Meilleure scalabilité).

MapReduce est le modèle de programmation (Framework) (legacy) de Hadoop 1.0/2.0 pour le calcul batch distribué (basé sur disque).

Il force le développeur à décomposer tout problème en 2 phases (fonctions) : Map et Reduce.

("le", 1)
("chat", 1)
("est", 1)
("sur", 1)
("le", 1)
("tapis", 1)

(Input pour la clé "le")
("le", [1, 1])

("le", 2)
("chat", 1)
("est", 1)
("sur", 1)
("tapis", 1)

Le Shuffle & Sort est l'étape (automatique, magique) qui se produit entre le Map et le Reduce. C'est le goulot d'étranglement (I/O) de MapReduce.

[ MAP 1 ]
("le", 1)
("chat", 1)
   │
   │ (Shuffle : Envoi "le" au Reducer 1)
   │ (Shuffle : Envoi "chat" au Reducer 2)
   ▼
[ MAP 2 ]
("le", 1)
("tapis", 1)
   │
   │ (Shuffle : Envoi "le" au Reducer 1)
   │ (Shuffle : Envoi "tapis" au Reducer 2)
   ▼
---------------- (RÉSEAU / DISQUE) ----------------
   │                                   │
   ▼                                   ▼
[ REDUCER 1 ] (Clé "le")            [ REDUCER 2 ] (Autres clés)
(Reçoit ("le", 1), ("le", 1))       (Reçoit ("chat", 1), ("tapis", 1))
(Tri (Sort) -> ("le", [1, 1]))      (Tri (Sort) -> ...)
   │                                   │
   ▼                                   ▼
("le", 2)                           ("chat", 1), ("tapis", 1)

Le "Shuffle" est le processus (coûteux) où les Mappers écrivent (sur disque local) leurs résultats (partitions), et où les Reducers lisent (via réseau) les données dont ils ont besoin depuis tous les Mappers.

"Word Count" (Compteur de Mots) est l'exemple "Hello, World!" du Big Data (MapReduce).

// 1. Le Mapper (Java)
public class WordCountMapper
    extends Mapper {
    
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    // (Appelé 1x par ligne)
    public void map(Object key, Text value, Context context)
        throws IOException, InterruptedException {
        
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            // (Output : ("mot", 1))
            context.write(word, one);
        }
    }
}

// 2. Le Reducer (Java)
public class WordCountReducer
    extends Reducer {
    
    private IntWritable result = new IntWritable();

    // (Appelé 1x par mot unique)
    // (ex: key="le", values=[1, 1])
    public void reduce(Text key, Iterable values, Context context)
        throws IOException, InterruptedException {
        
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        // (Output : ("le", 2))
        context.write(key, result);
    }
}

Problème : Écrire du Java MapReduce (5.3) est très difficile. Les analystes de données (BI) connaissent le SQL, pas le Java.

Solution : Apache Hive

Hive est un Data Warehouse (DWH) open-source (créé par Facebook) construit sur Hadoop.

Il fournit :

1. Un Metastore (Catalogue) : Une BDD (ex: MySQL) qui stocke les schémas (CREATE TABLE...) des données.
2. Un Driver (Moteur) : Un "traducteur" qui convertit le SQL (HiveQL) en Jobs MapReduce (ou Spark/Tez).

Flux (Hive)

1. [Analyste BI]
   │ (Écrit du SQL)
   │ "SELECT COUNT(*) FROM logs WHERE ... GROUP BY ..."
   │
   ▼
2. [Hive Server]
   │ (Parse le SQL (HiveQL))
   │ (Consulte le Metastore)
   │ (Génère un plan MapReduce/Spark)
   │
   ▼
3. [YARN Cluster]
   │ (Exécute le Job MapReduce/Spark)
   │ (Lit les données sur HDFS)
   │
   ▼
4. [Analyste BI] (Reçoit le résultat)

Usage : C'est l'ancêtre (et la base) de Spark SQL, Redshift Spectrum, BigQuery, etc.

Apache Pig (créé par Yahoo!) est une alternative à Java MapReduce (comme Hive).

Philosophie : Dataflow Language (Langage de Flux de Données).

Au lieu d'écrire du SQL (Hive) ou du Java (MapReduce), le développeur écrit un script Pig Latin (un langage de "pipeline" procédural).

Exemple (Pig Latin)

-- 1. Charger (Load) les données
logs = LOAD '/data/logs.txt' AS (ligne:chararray);

-- 2. "Tokenize" (Map)
mots = FOREACH logs GENERATE FLATTEN(TOKENIZE(ligne)) AS mot;

-- 3. Grouper (Shuffle/Reduce)
groupes = GROUP mots BY mot;

-- 4. Compter (Reduce)
comptes = FOREACH groupes GENERATE group AS mot, COUNT(mots) AS total;

-- 5. Sauvegarder
STORE comptes INTO '/data/output/wordcount';

Exécution : Le moteur "Pig" traduit ce script (Pig Latin) en un Job MapReduce (ou Spark) et l'exécute sur YARN.

Statut : Largement obsolète (remplacé par Spark DataFrames (PySpark)).

Apache HBase est une base de données NoSQL, distribuée, colonnaire (Wide-Column), construite sur HDFS. (Inspiré de Google Bigtable).

Hadoop (OLAP) vs HBase (OLTP)

Problème : Hadoop/HDFS/MapReduce sont "Batch" (Haute Latence). Ils ne peuvent pas faire : SELECT * FROM users WHERE id=1 (Accès Aléatoire Rapide).

Solution (HBase) : HBase est une BDD (non-SQL) qui tourne *sur* HDFS, mais qui fournit un accès Temps Réel (Faible Latence) (lecture/écriture) par Clé (RowKey).

Modèle (Wide-Column)

Optimisé pour des milliards de lignes et des millions de colonnes (dynamiques).

Théorème CAP

Dans le Théorème CAP (Consistency, Availability, Partition Tolerance), HBase choisit CP (Consistency & Partition Tolerance). (Il utilise ZooKeeper).

Apache Sqoop (SQL-to-Hadoop) est un outil (CLI) conçu pour transférer (ingérer) des données "Bulk" (massives) entre Hadoop (HDFS, Hive) et des bases de données relationnelles (RDBMS) (MySQL, Postgres, Oracle...).

sqoop import (RDBMS -> HDFS)

(Usage n°1) Ingestion (Batch) d'une BDD de production (OLTP) vers le Data Lake (OLAP).

# (Lance un job MapReduce (parallèle)
#  pour 'SELECT *' de 'ma_table_sql'
#  et l'écrire en 'Parquet' sur HDFS)
$ sqoop import \
    --connect jdbc:mysql://db.host/database \
    --username user --password pass \
    --table ma_table_sql \
    --target-dir /user/hive/warehouse/ma_table_hdfs \
    --as-parquetfile \
    -m 8  # (8 Mappers (parallèle))

sqoop export (HDFS -> RDBMS)

(Usage n°2) Exporter les résultats (ex: un rapport agrégé) de HDFS vers une table SQL.

Apache Flume est un service (distribué) conçu pour collecter, agréger, et déplacer (ingérer) des volumes massifs de logs (streaming).

Architecture (Agent)

Un "Agent" Flume (installé sur ex: un serveur web) est composé de 3 parties :

• Source : (L'entrée) "Où écouter ?" (ex: tail un fichier .log, écouter un port Syslog, un topic Kafka).
• Channel (Canal) : (La "Queue") Le "tampon" (buffer) de rétention (ex: MemoryChannel (rapide) ou FileChannel (fiable)).
• Sink (Destination) : (La sortie) "Où écrire ?" (ex: HDFS, HBase, Kafka, un autre Agent Flume).

(Ex: Agent 'web_server_logs')
Source (Tail /var/log/nginx/access.log)
   │
   ▼
Channel (Memory)
   │
   ▼
Sink (HDFS) -> (Écrit /logs/nginx/...)

Apache Oozie est un ordonnanceur (scheduler) de workflow (flux de travail) natif de l'écosystème Hadoop.

C'est l'équivalent (legacy) d'Airflow ou Azure Data Factory (Pipelines).

Usage (Gérer les DAGs)

Hadoop (MapReduce, Pig, Sqoop) est composé de "Jobs" indépendants. Oozie permet de les chaîner (créer un DAG).

Configuration (XML)

Les workflows Oozie sont (notoirement) complexes à écrire (XML).

<workflow-app name="Mon-Workflow-ETL" ...>
    <start to="ingestion_sqoop"/>
    
    <action name="ingestion_sqoop">
        <sqoop> ... (Action Sqoop) ... </sqoop>
        <ok to="transformation_pig"/>
        <error to="kill_job"/>
    </action>
    
    <action name="transformation_pig">
        <pig> ... (Action Pig) ... </pig>
        <ok to="end"/>
        <error to="kill_job"/>
    </action>
    
    <kill name="kill_job">...</kill>
    <end name="end"/>
</workflow-app>

Statut : Largement obsolète, remplacé par des orchestrateurs modernes (Airflow, Prefect, Dagster, ou (Databricks) Jobs Workflows).