Question

我的问题是关于mapreduce在java中的编程。

假设我有一个WordCount.java示例，一个标准的mapreduce程序。我希望map函数收集一些信息，并返回到reduce函数映射，形成如下：<；slaveNode_id、some_info_collected>，



以便我可以知道哪个从属节点收集了哪些数据。。知道怎么做吗？？

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）了解其运行的基础设施。

一般来说，答案是您的应用程序不需要这些信息。对Mapper的每次调用和对Reducer的每次调用都可以在不同的节点上执行，也可以全部在同一节点上执行。MapReduce的美妙之处在于结果是一样的，所以对于您的应用程序来说：这并不重要。

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

学习Hadoop玩得开心：）

P.S. The only way I can think of (which is nasty to say the least) is that you include a system call of some sort in the Mapper and ask the underlying OS about it s name/properties/etc. This kind of construct would make your application very non-portable; i.e. it won t run on Hadoop in Windows or Amazon.

Answer 2

字数对你来说是个错误的例子。您希望简单地将所有信息合并在一起。这就把事情倒过来算字数。

基本上，您只是将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)
  }
}

友情链接