hadoop
Introducción a MapReduce

Sintaxis

• Para ejecutar el ejemplo, la sintaxis del comando es:

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

• Para copiar datos en HDFS (desde local):

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

Observaciones

Programa de conteo de palabras (en Java y Python)

El programa de conteo de palabras es como el programa "Hello World" en MapReduce.

Hadoop MapReduce es un marco de software para escribir aplicaciones que procesan vastas cantidades de datos (conjuntos de datos de varios terabytes) en paralelo en grandes clusters (miles de nodos) de hardware básico de una manera confiable y tolerante a fallos.

Un trabajo MapReduce generalmente divide el conjunto de datos de entrada en fragmentos independientes que son procesados ​​por las tareas del mapa de una manera completamente paralela. El marco ordena los resultados de los mapas, que luego se ingresan en las tareas de reducción. Normalmente, tanto la entrada como la salida del trabajo se almacenan en un sistema de archivos. El marco se encarga de programar las tareas, monitorearlas y volver a ejecutar las tareas fallidas.

Ejemplo de conteo de palabras:

El ejemplo de WordCount lee archivos de texto y cuenta la frecuencia con la que aparecen las palabras. La entrada son archivos de texto y la salida son archivos de texto, cada línea de los cuales contiene una palabra y el conteo de la frecuencia con la que se produjo, separados por una pestaña.

Cada mapeador toma una línea como entrada y la divide en palabras. A continuación, emite un par de clave / valor de la palabra y cada reductor suma los conteos de cada palabra y emite una única clave / valor con la palabra y la suma.

Como optimización, el reductor también se utiliza como combinador en las salidas del mapa. Esto reduce la cantidad de datos enviados a través de la red al combinar cada palabra en un solo registro.

Código de conteo de palabras:

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);
 }
        
}

Para ejecutar el ejemplo, la sintaxis del comando es:

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

Se leen todos los archivos en el directorio de entrada (llamado in-dir en la línea de comando arriba) y los conteos de palabras en la entrada se escriben en el directorio de salida (llamado out-dir arriba). Se supone que tanto las entradas como las salidas se almacenan en HDFS. Si su entrada no está ya en HDFS, sino que se encuentra en un sistema de archivos local en algún lugar, debe copiar los datos en HDFS utilizando un comando como este:

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

Ejemplo de Word Count en 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)

El programa anterior se puede ejecutar usando cat filename.txt | python mapper.py | sort -k1,1 | python reducer.py