mapreduce 作业的链接答案

【问题标题】：Chaining of mapreduce jobsmapreduce 作业的链接
【发布时间】：2016-06-29 23:29:52
【问题描述】：

我遇到了“mapreduce 作业的链接”。作为 mapreduce 的新手，在什么情况下我们必须链接（我假设链接意味着依次运行 mapreduce 作业）作业？

有没有什么例子可以提供帮助？

【问题讨论】：

标签： hadoop mapreduce

【解决方案1】：

必须链接作业的经典示例是输出按频率排序的单词的字数统计。

你需要：

工作一：

输入源映射器（将单词作为键发出，一个作为值发出）
聚合reducer（聚合字数）

工作 2：

键/值交换映射器（以频率为键，词为值）
隐式身份归约器（获取按频率排序的单词，不必实现）

这是上面的映射器/减速器的例子：

public class HadoopWordCount {


  public static class TokenizerMapper extends Mapper<Object, Text, Text, LongWritable> {

    private final static Text word = new Text();
    private final static LongWritable one = new LongWritable(1);

    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  public static class KeyValueSwappingMapper extends Mapper<Text, LongWritable, LongWritable, Text> {

    public void map(Text key, LongWritable value, Context context) throws IOException, InterruptedException {
      context.write(value, key);
    }
  }

  public static class SumReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    private LongWritable result = new LongWritable();

    public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException,
        InterruptedException {
      long sum = 0;
      for (LongWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
}

这里是驱动程序的例子。

它需要两个参数：

一个用于计算字数的输入文本文件。
输出目录（不应预先存在） - 在 {this dir}/out2/part-r-0000 文件中查找输出

public static void main(String[] args) throws Exception {

    Configuration conf = new Configuration();
    Path out = new Path(args[1]);

    Job job1 = Job.getInstance(conf, "word count");
    job1.setJarByClass(HadoopWordCount.class);
    job1.setMapperClass(TokenizerMapper.class);
    job1.setCombinerClass(SumReducer.class);
    job1.setReducerClass(SumReducer.class);
    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(LongWritable.class);
    job1.setOutputFormatClass(SequenceFileOutputFormat.class);
    FileInputFormat.addInputPath(job1, new Path(args[0]));
    FileOutputFormat.setOutputPath(job1, new Path(out, "out1"));
    if (!job1.waitForCompletion(true)) {
      System.exit(1);
    }
    Job job2 = Job.getInstance(conf, "sort by frequency");
    job2.setJarByClass(HadoopWordCount.class);
    job2.setMapperClass(KeyValueSwappingMapper.class);
    job2.setNumReduceTasks(1);
    job2.setSortComparatorClass(LongWritable.DecreasingComparator.class);
    job2.setOutputKeyClass(LongWritable.class);
    job2.setOutputValueClass(Text.class);
    job2.setInputFormatClass(SequenceFileInputFormat.class);
    FileInputFormat.addInputPath(job2, new Path(out, "out1"));
    FileOutputFormat.setOutputPath(job2, new Path(out, "out2"));
    if (!job2.waitForCompletion(true)) {
      System.exit(1);
    }

}

【讨论】：

非常感谢尤吉斯。让我看看这个样本。亲切的问候。
为什么我们必须为job1设置OutputFormatClass和为job2设置InputFormatClass？如果没有这些语句，我会在 KeyValueSwappingMapper 中得到 ClassCastException。而且我不确定该异常到底是在哪里引发的。
非常感谢您的详细解答。

【解决方案2】：

简单地说，当您的问题不能仅适用于一个 map reduce 作业时，您必须链接多个 map reduce 作业。

一个很好的例子是找到前 10 名购买的物品，这可以通过 2 个工作来实现：

一个 map reduce 作业，用于查找每件商品的购买次数。
第二个工作，根据购买次数对物品进行排序，得到前10名的物品。

要获得完整的想法，作业链接会生成写入磁盘和从磁盘读取的中间文件，因此会降低性能。 尽量避免链接作业。

还有here 如何链接作业。

【讨论】：

您的链接现在已损坏。