mapreduce ejemplo recuento

Question

Mi pregunta es sobre mapreduce programming in java.

Supongamos que tengo el ejemplo de WordCount.java, un estándar mapreduce program. Quiero que la función de mapa recopile cierta información y vuelva a los mapas de funciones de reducción formados como: <slaveNode_id,some_info_collected>,

de modo que I can know what slave node collected what data ... ¿Alguna idea de cómo?

public class WordCount { 

    public static class Map extends MapReduceBase implements 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, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { 
     String line = value.toString(); 
     StringTokenizer tokenizer = new StringTokenizer(line); 
     while (tokenizer.hasMoreTokens()) { 
      word.set(tokenizer.nextToken()); 
      output.collect(word, one); 
     } 
     } 
    } 

    public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> { 
     public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { 
     int sum = 0; 
     while (values.hasNext()) { 
      sum += values.next().get(); 
     } 
     output.collect(key, new IntWritable(sum)); 
     } 
    } 

    public static void main(String[] args) throws Exception { 
     JobConf conf = new JobConf(WordCount.class); 
     conf.setJobName("wordcount"); 

     conf.setOutputKeyClass(Text.class); 
     conf.setOutputValueClass(IntWritable.class); 

     conf.setMapperClass(Map.class); 
     conf.setCombinerClass(Reduce.class); 
     conf.setReducerClass(Reduce.class); 

     conf.setInputFormat(TextInputFormat.class); 
     conf.setOutputFormat(TextOutputFormat.class); 

     FileInputFormat.setInputPaths(conf, new Path(args[0])); 
     FileOutputFormat.setOutputPath(conf, new Path(args[1])); 

     JobClient.runJob(conf); 
    } 
} 

Gracias!

Answer 1

Lo que se está preguntando es dejar que la aplicación (la función de reducir el mapa) conozca la infraestructura en la que se ejecutó.

En general la respuesta es que su aplicación no necesita esta información. Cada llamada al asignador y cada llamada al reductor se puede ejecutar en un nodo diferente o en el mismo nodo. La belleza de MapReduce es que el resultado es el mismo, por lo que para su aplicación: no importa.

Como consecuencia de la API no tienen características para apoyar esta solicitud de los suyos.

Diviértete aprendiendo Hadoop :)

---

P. S. La única forma en que puedo pensar (lo cual es desagradable por decir lo menos) es que incluye una llamada al sistema de algún tipo en el Mapper y le pregunta al sistema operativo subyacente sobre su nombre/propiedades/etc. Este tipo de construcción haría que su aplicación no fuera portátil; es decir, no se ejecutará en Hadoop en Windows o Amazon.

Answer 2

Número de palabras es un mal ejemplo para usted. Simplemente desea fusionar toda la información. Esto invierte las cosas a la palabra.

Básicamente, solo está emitiendo su slaveNode_id como IntWritable (si es posible) y la información como Text.

public static class Map extends MapReduceBase implements Mapper<LongWritable, Text,IntWritable, Text> { 
    private Text word = new Text(); 

    public void map(LongWritable key, Text value, OutputCollector<IntWritable, Text> output, Reporter reporter) throws IOException { 
    String line = value.toString(); 
    StringTokenizer tokenizer = new StringTokenizer(line); 
    while (tokenizer.hasMoreTokens()) { 
     word.set(tokenizer.nextToken()); 
     // you have to split your data here: ID and value 
     IntWritable id = new IntWritable(YOUR_ID_HERE); 

     output.collect(id, word); 
    } 
    } 
} 

y el reductor podría seguir el mismo camino:

public static class Reduce extends MapReduceBase implements Reducer<IntWritable, Text,IntWritable, Text> { 
    public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable,Text> output, Reporter reporter) throws IOException { 

     // now you have all the values for a slaveID as key. Do whatever you like with that... 
     for(Text value : values) 
     output.collect(key, value) 
    } 
}