Hadoop mapreduce：用于链接MapReduce作业内映射器的驱动程序 [英] Hadoop mapreduce : Driver for chaining mappers within a MapReduce job

问题描述

我有mapreduce作业：
我的代码Map类：

  public static class MapClass extends Mapper< Text，文本，文本，LongWritable> {b 
 $ b @Override 
 public void map（Text key，Text value，Context context）
 throws IOException，InterruptedException {
} 
} 
  

我想使用ChainMapper：

  1。工作职位=新职位（conf，有链接任务的工作）; 
 2. job.setJarByClass（MapReduce.class）; 
 3. job.setInputFormatClass（TextInputFormat.class）; 
 4. job.setOutputFormatClass（TextOutputFormat.class）; 
 
 5. FileInputFormat.setInputPaths（job，new Path（InputFile））; 
 6. FileOutputFormat.setOutputPath（job，new Path（OutputFile））; 
 
 7. JobConf map1 = new JobConf（false）; 
 
 8. ChainMapper.addMapper（
 job，
 MapClass.class，
 Text.class，
 Text.class，
 Text。 class，
 Text.class，
 true，
 map1 
）; 
  

但其报告在第8行有错误：

此行的多个标记
- 发生'addMapper'
- 方法addMapper（JobConf，Class>，Class，Class，
Class， （Job，$ b $ Class，Class，Class，Class，Class，Class，boolean，Configuration）
- 调试当前指令指针
- 类型ChainMapper中的方法addMapper（JobConf，Class>，Class，Class，
Class，Class，boolean，JobConf）不适用于参数
（JobConf，Class，Class，Class，Class ，Class，boolean，JobConf）

解决方案

经过很多功夫能够使用 ChainMapper / ChainReducer 。感谢您的最新评论 user864846。

  / ** 
 *授权给Apache软件基金会（ASF）根据
 *或更多的贡献者许可协议。请参阅此作品发行的NOTICE文件
 *以了解有关版权所有权的其他信息
 *。 ASF根据Apache许可证2.0版（
 *许可证）将此文件
 *许可给您;除了符合
 *的许可证外，您不得使用此文件。您可以在
 * 
 * http://www.apache.org/licenses/LICENSE-2.0 
 * 
 *处获得许可证副本*除非适用法律要求或以书面形式同意，根据许可证分发的软件
 *以原样基础，
 *分发，不附有任何明示或暗示的保证或条件。 
 *请参阅许可证以了解许可证下特定语言的管理权限和
 *限制。 
 * / 
 
包myPKG; 
 
 / * 
 * Ajitsen：ChainMapper / ChainReducer的示例程序。该程序是Hadoop-0.18.0中提供的WordCount示例的修改版本。添加了ChainMapper / ChainReducer，并在Hadoop 1.0.2中工作。 
 * / 
 
 import java.io.IOException; 
 import java.util.Iterator; 
 import java.util.StringTokenizer; 
 
导入org.apache.hadoop.conf.Configuration; 
导入org.apache.hadoop.conf.Configured; 
导入org.apache.hadoop.fs.Path; 
 import org.apache.hadoop.io.IntWritable; 
 import org.apache.hadoop.io.LongWritable; 
 import org.apache.hadoop.io.Text; 
 import org.apache.hadoop.mapred。*; 
 import org.apache.hadoop.mapred.lib.ChainMapper; 
 import org.apache.hadoop.mapred.lib.ChainReducer; 
 import org.apache.hadoop.util.Tool; 
 import org.apache.hadoop.util.ToolRunner; 
 
 public class ChainWordCount extends Configured implements Tool {
 
 public static class Tokenizer extends MapReduceBase 
 implements Mapper< LongWritable，Text，Text，IntWritable> {
 
 private static static IntWritable one = new IntWritable（1）; 
私人文字=新文字（）; 
 $ b $ public void map（LongWritable key，Text value，
 OutputCollector< Text，IntWritable> output，
 Reporter reporter）throws IOException {
 String line = value.toString （）; 
 System.out.println（Line：+ line）; 
 StringTokenizer itr = new StringTokenizer（line）; 
 while（itr.hasMoreTokens（））{
 word.set（itr.nextToken（））; 
 output.collect（word，one）; 
 
 
 
 $ b public static class UpperCaser extends MapReduceBase 
 implements Mapper< Text，IntWritable，Text，IntWritable> {
 $ b $ public void map（Text key，IntWritable value，
 OutputCollector< Text，IntWritable> output，
 Reporter reporter）throws IOException {
 String word = key。的toString（）toUpperCase（）; 
 System.out.println（大写：+单词）; 
 output.collect（new Text（word），value）; 
 
 
 
 public static class Reduce extends MapReduceBase 
 implements Reducer< Text，IntWritable，Text，IntWritable> {
 $ b $ 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（）; 
} 
 System.out.println（Word：+ key.toString（）+\tCount：+ sum）; 
 output.collect（key，new IntWritable（sum））; 
 
 
 
 static int printUsage（）{
 System.out.println（wordcount< input>< output>）; 
 ToolRunner.printGenericCommandUsage（System.out）; 
返回-1; 
} 
 
 public int run（String [] args）throws Exception {
 JobConf conf = new JobConf（getConf（），ChainWordCount.class）; 
 conf.setJobName（wordcount）; 
 
 if（args.length！= 2）{
 System.out.println（错误：错误的参数数目：+ 
 args.length +而不是2 。）; 
 return printUsage（）; 
} 
 FileInputFormat.setInputPaths（conf，args [0]）; 
 FileOutputFormat.setOutputPath（conf，new Path（args [1]））; 
 
 conf.setInputFormat（TextInputFormat.class）; 
 conf.setOutputFormat（TextOutputFormat.class）; 
 
 JobConf mapAConf = new JobConf（false）; 
 ChainMapper.addMapper（conf，Tokenizer.class，LongWritable.class，Text.class，Text.class，IntWritable.class，true，mapAConf）; 
 
 JobConf mapBConf = new JobConf（false）; 
 ChainMapper.addMapper（conf，UpperCaser.class，Text.class，IntWritable.class，Text.class，IntWritable.class，true，mapBConf）; 
 
 JobConf reduceConf = new JobConf（false）; 
 ChainReducer.setReducer（conf，Reduce.class，Text.class，IntWritable.class，Text.class，IntWritable.class，true，reduceConf）; 
 
 JobClient.runJob（conf）; 
返回0; 
 
 $ b $ public static void main（String [] args）throws Exception {
 int res = ToolRunner.run（new Configuration（），new ChainWordCount（），args）; 
 System.exit（res）; 
 
 
 $ / code $ / pre 
 
 $ hr 
 
 
最新版本的 EDIT （至少从），addMapper中的 true 标志不是必需的。 （实际上签名有变化抑制它）。
 
 
 
所以它只是 
 
 
  JobConf mapAConf = new JobConf（false）; 
 ChainMapper.addMapper（conf，Tokenizer.class，LongWritable.class，Text.class，
 Text.class，IntWritable.class，mapAConf）; 
  
 
I have mapreduce job:
my code Map class:
public static class MapClass extends Mapper<Text, Text, Text, LongWritable> {

    @Override
    public void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {
    }
}
And I want to use ChainMapper :
1. Job job = new Job(conf, "Job with chained tasks");
2. job.setJarByClass(MapReduce.class);
3. job.setInputFormatClass(TextInputFormat.class);
4. job.setOutputFormatClass(TextOutputFormat.class);

5. FileInputFormat.setInputPaths(job, new Path(InputFile));
6. FileOutputFormat.setOutputPath(job, new Path(OutputFile));

7. JobConf map1 = new JobConf(false);

8. ChainMapper.addMapper(
        job, 
        MapClass.class, 
        Text.class, 
        Text.class, 
        Text.class, 
        Text.class, 
        true, 
        map1
        ); 
but its report has an error at line 8 :

  
Multiple markers at this line
      - Occurrence of 'addMapper'
      - The method addMapper(JobConf, Class>, Class, Class, 
       Class, Class, boolean, JobConf) in the type ChainMapper is not applicable for the arguments (Job, 
       Class, Class, Class, Class, Class, boolean, Configuration)
      - Debug Current Instruction Pointer
      - The method addMapper(JobConf, Class>, Class, Class, 
       Class, Class, boolean, JobConf) in the type ChainMapper is not applicable for the arguments 
       (JobConf, Class, Class, Class, Class, Class, boolean, JobConf)

解决方案
After a lot of "Kung Fu", I was able to use ChainMapper/ChainReducer. Thanks for last comment user864846.
/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package myPKG;

/* 
 * Ajitsen: Sample program for ChainMapper/ChainReducer. This program is modified version of WordCount example available in Hadoop-0.18.0. Added ChainMapper/ChainReducer and made to works in Hadoop 1.0.2. 
 */

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.ChainMapper;
import org.apache.hadoop.mapred.lib.ChainReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ChainWordCount extends Configured implements Tool {

    public static class Tokenizer 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();
            System.out.println("Line:"+line);
            StringTokenizer itr = new StringTokenizer(line);
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                output.collect(word, one);
            }
        }
    }

    public static class UpperCaser extends MapReduceBase
    implements Mapper<Text, IntWritable, Text, IntWritable> {

        public void map(Text key, IntWritable value, 
                OutputCollector<Text, IntWritable> output, 
                Reporter reporter) throws IOException {
            String word = key.toString().toUpperCase();
            System.out.println("Upper Case:"+word);
            output.collect(new Text(word), value);    
        }
    }

    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();
            }
            System.out.println("Word:"+key.toString()+"\tCount:"+sum);
            output.collect(key, new IntWritable(sum));
        }
    }

    static int printUsage() {
        System.out.println("wordcount <input> <output>");
        ToolRunner.printGenericCommandUsage(System.out);
        return -1;
    }

    public int run(String[] args) throws Exception {
        JobConf conf = new JobConf(getConf(), ChainWordCount.class);
        conf.setJobName("wordcount");

        if (args.length != 2) {
            System.out.println("ERROR: Wrong number of parameters: " +
                    args.length + " instead of 2.");
            return printUsage();
        }
        FileInputFormat.setInputPaths(conf, args[0]);
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));

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

        JobConf mapAConf = new JobConf(false);
        ChainMapper.addMapper(conf, Tokenizer.class, LongWritable.class, Text.class, Text.class, IntWritable.class, true, mapAConf);

        JobConf mapBConf = new JobConf(false);
        ChainMapper.addMapper(conf, UpperCaser.class, Text.class, IntWritable.class, Text.class, IntWritable.class, true, mapBConf);

        JobConf reduceConf = new JobConf(false);
        ChainReducer.setReducer(conf, Reduce.class, Text.class, IntWritable.class, Text.class, IntWritable.class, true, reduceConf);

        JobClient.runJob(conf);
        return 0;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new ChainWordCount(), args);
        System.exit(res);
    }
}




EDIT in latest version (at least from hadoop 2.6), the true flag in addMapper is not needed. (in fact the signature has change suppression it`).

So it would be just
JobConf mapAConf = new JobConf(false);
ChainMapper.addMapper(conf, Tokenizer.class, LongWritable.class, Text.class,
                      Text.class, IntWritable.class, mapAConf);


                        
这篇关于Hadoop mapreduce：用于链接MapReduce作业内映射器的驱动程序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！