mapreduce计数示例

Question

我的问题是关于mapreduce programming in java。

假设我有WordCount.java示例，标准为mapreduce program。我想要地图功能收集一些信息，并返回到缩小功能地图形成如：<slaveNode_id,some_info_collected>,

这样I can know what slave node collected what data ..任何想法如何？

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

谢谢！

Answer 1

你问的是让应用程序（你的map-reduce thingy）知道它运行的基础设施。

一般来说，答案是您的应用程序不需要此信息。对映射器的每次调用和对Reducer的每次调用都可以在不同的节点上执行，也可以在同一节点上执行。 MapReduce的优点在于结果是一样的，所以对于你的应用程序来说：没关系。

因此，API没有支持您的请求的功能。

玩得开心学习Hadoop的:)

---

附：我能想到的唯一方法（至少可以说是令人讨厌的）是你在Mapper中包含了某种类型的系统调用，并询问底层操作系统是否有名称/属性等。 这种构造会使你的应用程序非常不便携;即它不会在Windows或亚马逊的Hadoop上运行。

Answer 2

Wordcount是你的错误例子。你想简单地将所有信息合并在一起。这反过来的事情wordcount。

基本上，你只是发出你的slaveNode_id作为IntWritable（如果这是可能的话）和信息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); 
    } 
    } 
} 

和减速器会走一样的路：

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