hadoop
Introduzione a MapReduce

Sintassi

• Per eseguire l'esempio, la sintassi del comando è:

  bin/hadoop jar hadoop-*-examples.jar wordcount [-m <#maps>] [-r <#reducers>] <in-dir> <out-dir>
  

• Per copiare i dati in HDFS (dal locale):

  bin/hadoop dfs -mkdir <hdfs-dir> //not required in hadoop 0.17.2 and later
  bin/hadoop dfs -copyFromLocal <local-dir> <hdfs-dir>
  

Osservazioni

Word Count Program (in Java e Python)

Il programma di conteggio delle parole è come il programma "Hello World" in MapReduce.

Hadoop MapReduce è un framework software per la scrittura semplice di applicazioni che elaborano grandi quantità di dati (set di dati multi terabyte) in parallelo su cluster di grandi dimensioni (migliaia di nodi) di hardware di base in modo affidabile e tollerante ai guasti.

Un lavoro MapReduce solitamente divide il set di dati di input in blocchi indipendenti che vengono elaborati dalle attività della mappa in modo completamente parallelo. Il framework ordina le uscite delle mappe, che vengono poi inserite nelle attività di riduzione. In genere, sia l'input che l'output del lavoro sono memorizzati in un file system. Il framework si occupa delle attività di pianificazione, monitorandole e rielabora le attività fallite.

Esempio di conteggio parole:

L'esempio di WordCount legge i file di testo e conta quante volte si verificano le parole. L'input è costituito da file di testo e l'output è costituito da file di testo, ciascuna delle quali contiene una parola e il conteggio della frequenza con cui si è verificato, separati da una scheda.

Ogni mappatore prende una riga come input e la spezza in parole. Quindi emette una coppia chiave / valore della parola e ciascun riduttore somma i conteggi per ogni parola ed emette una singola chiave / valore con la parola e la somma.

Come ottimizzazione, il riduttore viene anche utilizzato come combinatore sulle uscite della mappa. Ciò riduce la quantità di dati inviati attraverso la rete combinando ogni parola in un singolo record.

Codice di conteggio parole:

package org.myorg;
        
import java.io.IOException;
import java.util.*;
        
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
        
public class WordCount {
        
 public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
        
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
 } 
        
 public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) 
      throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
 }
        
 public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
        
        Job job = new Job(conf, "wordcount");
    
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
        
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);
        
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
        
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
    job.waitForCompletion(true);
 }
        
}

Per eseguire l'esempio, la sintassi del comando è:

bin/hadoop jar hadoop-*-examples.jar wordcount [-m <#maps>] [-r <#reducers>] <in-dir> <out-dir>

Tutti i file nella directory di input (chiamati in-dir nella riga di comando precedente) vengono letti e i conteggi delle parole nell'input vengono scritti nella directory di output (chiamata out-dir sopra). Si presume che sia gli input che gli output siano memorizzati in HDFS. Se il tuo input non è già in HDFS, ma è piuttosto in un file system locale da qualche parte, devi copiare i dati in HDFS usando un comando come questo:

bin/hadoop dfs -mkdir <hdfs-dir> //not required in hadoop 0.17.2 and later
bin/hadoop dfs -copyFromLocal <local-dir> <hdfs-dir>

Esempio di conteggio parole in Python:

mapper.py

import sys 
for line in sys.stdin: 
    # remove leading and trailing whitespace 
    line = line.strip() 
    # split the line into words 
    words = line.split() 
    # increase counters 
    for word in words: 
        print '%s\t%s' % (word, 1)

reducer.py

import sys
current_word = None
current_count = 0
word = None
for line in sys.stdin:
    # remove leading and trailing whitespaces
    line = line.strip()
    # parse the input we got from mapper.py
    word, count = line.split('\t', 1)
    # convert count (currently a string) to int
    try:
        count = int(count)
    except ValueError:
        # count was not a number, so silently
        # ignore/discard this line
        continue
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print '%s\t%s' % (current_word, current_count)
        current_count = count
        current_word = word
if current_word == word:
    print '%s\t%s' % (current_word, current_count)

Il programma sopra può essere eseguito usando cat filename.txt | python mapper.py | sort -k1,1 | python reducer.py