La data,
en grand format.

┌──────────────────────────────────────────┐
│              NAMENODE (Maître)            │
│  - Métadonnées (noms, emplacements)      │
│  - Gère l'arbre du système de fichiers   │
│  - NE stocke PAS les données             │
└──────────┬──────────┬───────────┬────────┘
           │          │           │
     ┌─────▼──┐  ┌────▼───┐  ┌───▼────┐
     │DataNode│  │DataNode│  │DataNode│
     │ Bloc A │  │ Bloc B │  │ Bloc A │  ← Réplication
     │ Bloc C │  │ Bloc A │  │ Bloc B │
     └────────┘  └────────┘  └────────┘

Données d'entrée → [SPLIT] → MAP → [SHUFFLE & SORT] → REDUCE → Résultat

hdfs namenode -format      # Formater le NameNode (une seule fois !)
start-dfs.sh               # Démarrer HDFS
start-yarn.sh              # Démarrer YARN
jps                        # Vérifier les processus

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WCount {
    public static void main(String[] args) throws Exception {
        // 1. Créer la configuration
        Configuration conf = new Configuration();

        // 2. Récupérer les arguments
        String[] ourArgs =
            new GenericOptionsParser(conf, args).getRemainingArgs();

        // 3. Créer le Job
        Job job = Job.getInstance(conf, "Compteur de mots v1.0");

        // 4. Définir les classes
        job.setJarByClass(WCount.class);          // Driver
        job.setMapperClass(WCountMap.class);       // MAP
        job.setReducerClass(WCountReduce.class);   // REDUCE

        // 5. Définir les types de sortie
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 6. Définir les entrées/sorties HDFS
        FileInputFormat.addInputPath(job, new Path(ourArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(ourArgs[1]));

        // 7. Exécuter et attendre
        if (job.waitForCompletion(true))
            System.exit(0);
        System.exit(-1);
    }
}

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WCountMap extends Mapper<Object, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    protected void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        // Découper la ligne en mots
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            // Émettre (mot, 1)
            context.write(word, one);
        }
    }
}

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WCountReduce extends Reducer<Text, IntWritable, Text, Text> {

    protected void reduce(Text key, Iterable<IntWritable> values,
            Context context) throws IOException, InterruptedException {
        int sum = 0;
        // Sommer toutes les occurrences
        for (IntWritable val : values) {
            sum += val.get();
        }
        // Émettre (mot, total)
        context.write(key, new Text(String.valueOf(sum)));
    }
}

# Compiler
hadoop com.sun.tools.javac.Main WCount*.java

# Créer le JAR
mkdir -p build
mv WCount*.class build
jar -cvf wc.jar -C build .

# Mettre les données sur HDFS
hdfs dfs -put poeme.txt data/

# Exécuter
hadoop jar wc.jar WCount data results/wcoutput

# Voir les résultats
hdfs dfs -ls results/wcoutput
hdfs dfs -cat results/wcoutput/part-r-00000

#!/usr/bin/env python
import sys

for line in sys.stdin:
    line = line.strip()
    words = line.split()
    for word in words:
        print('%s\t%s' % (word, 1))

#!/usr/bin/env python
import sys

current_word = None
current_count = 0

for line in sys.stdin:
    line = line.strip()
    word, count = line.split('\t', 1)
    count = int(count)

    if current_word == word:
        current_count += count
    else:
        if current_word:
            print('%s\t%s' % (current_word, current_count))
        current_word = word
        current_count = count

if current_word == word:
    print('%s\t%s' % (current_word, current_count))

cat poeme.txt | python WordCount_mapper.py
cat poeme.txt | python WordCount_mapper.py | sort -k1,1 | python WordCount_reducer.py

# Trouver le JAR streaming
locate hadoop-streaming | grep jar

# Exécuter
hadoop jar /opt/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.9.2.jar \
    -input data/poeme.txt \
    -output results/outputpython \
    -mapper "python WordCount_mapper.py" \
    -reducer "python WordCount_reducer.py" \
    -file HadoopStreaming/WordCount_mapper.py \
    -file HadoopStreaming/WordCount_reducer.py

# Résultats
hdfs dfs -cat results/outputpython/part-00000

┌─────────────────┐
│  Driver Program  │  ← Contient le SparkContext
│  (SparkContext)  │
└────────┬────────┘
         │ se connecte
   ┌─────▼──────┐
   │  Cluster   │  ← Spark Master / YARN / Mesos
   │  Manager   │
   └──┬────┬──┬─┘
      │    │  │
  ┌───▼┐ ┌▼──▼┐
  │Exec│ │Exec│  ← Executors (sur les nœuds)
  │Task│ │Task│     Lancent les tâches + stockent les données
  └────┘ └────┘

val x = 10       // Immuable (comme final en Java)
var y = 20       // Mutable (modifiable)
val z: Int = 30  // Avec type explicite

// Fonction nommée
def add(a: Int, b: Int): Int = a + b

// Fonction anonyme (lambda)
val double = (x: Int) => x * 2

// Fonction sans paramètre
def hello() = println("Hello!")

val l1 = List(1, 2, 3)        // Immuable
val l2 = 0 :: l1               // Ajout en tête → List(0,1,2,3)
val l3 = l1 :+ 4               // Ajout en fin → List(1,2,3,4)
val l4 = l1 ::: l3             // Concaténation

val m1 = Map("A" -> 1, "B" -> 10, "C" -> 30)
m1("B")         // 10
m1.get("F")     // None (pas d'exception !)
m1 + ("D" -> 50) // Nouvelle map avec D ajouté

val s1 = Set(1, 2, 3, 4, 5)   // Immuable
val s2 = s1 + 6                // Ajout
val s3 = s1 - 2                // Suppression
s3.contains(3)                 // true

list.map(f)       // Applique f à chaque élément
list.filter(f)    // Garde les éléments où f retourne true
list.flatMap(f)   // map + aplatit le résultat
list.reduce(f)    // Agrège tous les éléments avec f
list.foldLeft(init)(f)  // Comme reduce mais avec valeur initiale
list.foreach(f)   // Applique f sans retour
list.flatten()    // Retire un niveau d'imbrication

// Classe
class Shape(cx: Int, cy: Int) {
    val x = cx
    val y = cy
    override def toString() = "(" + x + ", " + y + ")"
}

// Héritage
class Circle(cx: Int, cy: Int, cr: Int) extends Shape(cx, cy) {
    private var theRadius = Math.max(cr, 0)
    def radius = theRadius
}

// Objet Singleton (≈ static en Java)
object Bar {
    val x = 0
    def doSomething() { /* ... */ }
}

val m = Map("A" -> 10, "B" -> 20)
m.get("A")           // Some(10)
m.get("C")           // None
m.get("C").orElse(50) // 50 si absent

# Python
x = sc.parallelize(["b", "a", "c"])
y = x.map(lambda z: (z, 1))
# y: [('b',1), ('a',1), ('c',1)]

// Scala
val x = sc.parallelize(Array("b", "a", "c"))
val y = x.map(z => (z, 1))

x = sc.parallelize([1, 2, 3])
y = x.filter(lambda x: x % 2 == 1)
# y: [1, 3]

val x = sc.parallelize(Array(1, 2, 3))
val y = x.filter(n => n % 2 == 1)

x = sc.parallelize([1, 2, 3])
y = x.flatMap(lambda x: (x, x*100, 42))
# y: [1, 100, 42, 2, 200, 42, 3, 300, 42]

val x = sc.parallelize(Array(1, 2, 3))
val y = x.flatMap(n => Array(n, n*100, 42))

x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])
y = x.groupBy(lambda w: w[0])
# y: [('J',['John','James']), ('F',['Fred']), ('A',['Anna'])]

x = sc.parallelize([('B',5),('B',4),('A',3),('A',2),('A',1)])
y = x.groupByKey()
# y: [('A',[3,2,1]), ('B',[5,4])]

words = ["one", "two", "two", "three", "three", "three"]
wordPairsRDD = sc.parallelize(words).map(lambda word: (word, 1))
result = wordPairsRDD.reduceByKey(lambda a, b: a + b)
# result: [('one',1), ('two',2), ('three',3)]

val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))
val result = wordPairsRDD.reduceByKey(_ + _)

x = sc.parallelize([1, 2, 3, 4, 5])
y = x.sample(False, 0.4, 42)

x = sc.parallelize([1, 2, 3])
y = sc.parallelize([3, 4])
z = x.union(y)
# z: [1, 2, 3, 3, 4]

x = sc.parallelize([("a", 1), ("b", 2)])
y = sc.parallelize([("a", 3), ("a", 4), ("b", 5)])
z = x.join(y)
# z: [('a',(1,3)), ('a',(1,4)), ('b',(2,5))]

x = sc.parallelize([1, 2, 3, 3, 4])
y = x.distinct()
# y: [1, 2, 3, 4]

x = sc.parallelize([1, 2, 3, 4, 5], 3)
y = x.coalesce(2)

x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])
y = x.keyBy(lambda w: w[0])
# y: [('J','John'), ('F','Fred'), ('A','Anna'), ('J','James')]

x = sc.parallelize([1, 2, 3])
y = x.map(lambda n: n*n)
z = x.zip(y)
# z: [(1,1), (2,4), (3,9)]

x = sc.parallelize([1, 2, 3])
y = x.collect()
# y: [1, 2, 3]

x = sc.parallelize([1, 2, 3])
y = x.count()   # y: 3

x = sc.parallelize([1, 2, 3, 4])
y = x.reduce(lambda a, b: a + b)
# y: 10

seqOp = lambda data, item: (data[0] + [item], data[1] + item)
combOp = lambda d1, d2: (d1[0] + d2[0], d1[1] + d2[1])
x = sc.parallelize([1, 2, 3, 4])
y = x.aggregate(([], 0), seqOp, combOp)
# y: ([1, 2, 3, 4], 10)

x = sc.parallelize([2, 4, 1])
x.max()    # 4
x.sum()    # 7
x.mean()   # 2.333...
x.stdev()  # 1.247...

x = sc.parallelize([('J','James'), ('F','Fred'), ('A','Anna'), ('J','John')])
y = x.countByKey()
# y: {'A': 1, 'J': 2, 'F': 1}

x = sc.parallelize([2, 4, 1])
x.saveAsTextFile("/home/hduser/demo")

x = sc.parallelize([1, 2, 3, 4])
x.first()   # 1
x.take(2)   # [1, 2]

x = sc.parallelize([1, 2, 3], 2)
y = x.getNumPartitions()   # 2

# 1. Java JDK 8
wget https://github.com/frekele/oracle-java/releases/download/8u212-b10/jdk-8u212-linux-x64.tar.gz
tar zxvf jdk-8u212-linux-x64.tar.gz
su
mv jdk1.8.0_212 /usr/java

# 2. Créer l'utilisateur hadoop
groupadd hadoop
adduser -G hadoop hduser
passwd hduser
su - hduser

# 3. SSH sans mot de passe
ssh-keygen -t rsa -P ""
cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys
chmod 0600 ~/.ssh/authorized_keys
ssh localhost    # Tester
exit

# 4. Hadoop
wget https://archive.apache.org/dist/hadoop/core/hadoop-2.9.2/hadoop-2.9.2.tar.gz
tar xzf hadoop-2.9.2.tar.gz
su
mv hadoop-2.9.2 /opt/hadoop
su - hduser

# 5. Variables d'environnement (~/.bashrc)
export JAVA_HOME=/usr/java
export PATH=$PATH:$JAVA_HOME/bin
export HADOOP_HOME=/opt/hadoop
export HADOOP_INSTALL=$HADOOP_HOME
export HADOOP_MAPRED_HOME=$HADOOP_HOME
export HADOOP_COMMON_HOME=$HADOOP_HOME
export HADOOP_HDFS_HOME=$HADOOP_HOME
export YARN_HOME=$HADOOP_HOME
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export HADOOP_OPTS="-Djava.library.path=$HADOOP_INSTALL/lib/native"
export HADOOP_CLASSPATH=$JAVA_HOME/jre/lib:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar
export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin
source ~/.bashrc

# 6. Configurer hadoop-env.sh
export JAVA_HOME=/usr/java

# 7. Créer répertoires HDFS + configurer XML (voir chapitre 2)
mkdir -p /opt/hadoop/hadoopdata/namenode
mkdir -p /opt/hadoop/hadoopdata/datanode
chown -R hduser:hadoop /opt/hadoop

# 8. Formater & Démarrer
hdfs namenode -format
start-dfs.sh
start-yarn.sh
jps

# 9. Créer répertoires HDFS
hdfs dfs -mkdir /user
hdfs dfs -mkdir /user/hduser
hdfs dfs -chown hduser /user/hduser
hdfs dfs -mkdir data
hdfs dfs -mkdir results

# 1. Placer WCount.java, WCountMap.java, WCountReduce.java dans ~/WordCountJava
# 2. Compiler
hadoop com.sun.tools.javac.Main WordCountJava/WCount*.java
mkdir -p WordCountJava/build
mv WordCountJava/WCount*.class WordCountJava/build
jar -cvf wc.jar -C WordCountJava/build .

# 3. Préparer les données
hdfs dfs -put WordCountJava/poeme.txt data

# 4. Exécuter
hadoop jar wc.jar WCount data results/wcoutput

# 5. Résultats
hdfs dfs -ls results/wcoutput
hdfs dfs -cat results/wcoutput/part-r-00000

# Scala
wget https://downloads.lightbend.com/scala/2.12.8/scala-2.12.8.tgz
tar xvf scala-2.12.8.tgz
su
mv scala-2.12.8 /usr/scala
exit
# Ajouter à .bashrc :
export PATH=$PATH:/usr/scala/bin
source .bashrc
scala -version

# Spark
wget https://archive.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
tar xvf spark-2.4.5-bin-hadoop2.7.tgz
su
mv spark-2.4.5-bin-hadoop2.7 /usr/spark
exit
# Ajouter à .bashrc :
export PATH=$PATH:/usr/spark/bin
source .bashrc

# Tester
spark-shell       # Scala (Ctrl+D pour quitter)
pyspark           # Python (Ctrl+D pour quitter)

lines = sc.textFile('/home/haduser/data/poeme.txt')
words = lines.flatMap(lambda x: x.split(" "))
tuples = words.map(lambda x: (x, 1))
res = tuples.reduceByKey(lambda a, b: a + b)
res.saveAsTextFile('/home/haduser/results/pythonout')

val lines = sc.textFile("/home/haduser/data/poeme.txt")
val words = lines.flatMap(line => line.split(" "))
val tuples = words.map(word => (word, 1))
val res = tuples.reduceByKey(_ + _)
res.saveAsTextFile("/home/haduser/results/scalaout")

#!/usr/bin/env python
import sys
for line in sys.stdin:
    data = line.strip().split("\t")
    if len(data) == 6:
        store = data[2]
        cost = data[4]
        print('%s\t%s' % (store, cost))

#!/usr/bin/env python
import sys
current_store = None
current_total = 0
for line in sys.stdin:
    line = line.strip()
    store, cost = line.split('\t', 1)
    cost = float(cost)
    if current_store == store:
        current_total += cost
    else:
        if current_store:
            print('%s\t%.2f' % (current_store, current_total))
        current_store = store
        current_total = cost
if current_store:
    print('%s\t%.2f' % (current_store, current_total))

lines = sc.textFile("data/purchases.txt")
sales = lines.map(lambda l: l.split("\t")).map(lambda d: (d[2], float(d[4])))
result = sales.reduceByKey(lambda a, b: a + b)
result.collect()

# Mapper : extraire (produit, coût)
# clé = data[3] (produit), valeur = data[4] (coût)

# Spark :
lines = sc.textFile("data/purchases.txt")
sales = lines.map(lambda l: l.split("\t")).map(lambda d: (d[3], float(d[4])))
result = sales.reduceByKey(lambda a, b: a + b)
result.collect()
# Pour Toys et Consumer Electronics :
result.filter(lambda x: x[0] in ["Toys", "Consumer Electronics"]).collect()

# Mapper : extraire (magasin, coût)
# Reducer : garder le max au lieu de la somme

# Spark :
lines = sc.textFile("data/purchases.txt")
sales = lines.map(lambda l: l.split("\t")).map(lambda d: (d[2], float(d[4])))
result = sales.reduceByKey(lambda a, b: max(a, b))
# Pour Reno, Toledo, Chandler :
result.filter(lambda x: x[0] in ["Reno", "Toledo", "Chandler"]).collect()

# Spark :
lines = sc.textFile("data/purchases.txt")
costs = lines.map(lambda l: l.split("\t")).map(lambda d: float(d[4]))
total_count = costs.count()
total_value = costs.sum()
print("Nombre total:", total_count)
print("Valeur totale:", total_value)

# Mapper : extraire (jour_semaine, coût)
# Reducer : calculer la moyenne

# Spark :
lines = sc.textFile("data/purchases.txt")
sales = lines.map(lambda l: l.split("\t")).map(lambda d: (d[0], (float(d[4]), 1)))
sums = sales.reduceByKey(lambda a, b: (a[0]+b[0], a[1]+b[1]))
averages = sums.mapValues(lambda v: v[0]/v[1])
averages.collect()
# Pour Sunday :
averages.filter(lambda x: x[0] == "Sunday").collect()