eclipse - 在Mapreduce中做job chaining时，如何解决chainmapper is not applicable for the arguments错误？

Question

我正在使用 Hadoop 1.2.1，eclipse juno。我正在尝试在单个 Mapreduce 作业中链接三个 map task 。在 Eclipse 中编写 Mapreduce 代码时，出现错误，例如 chainmapper 不适用于参数，而且我无法设置输入路径。以下是我的 mapreduce 代码，

 package org.myorg;

import java.io.IOException;
import java.net.URI;
import java.nio.file.FileSystem;
import java.util.StringTokenizer;

import javax.security.auth.login.Configuration;

import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.MapRunnable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.ChainMapper;
import org.apache.hadoop.mapred.lib.ChainReducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.net.StaticMapping;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class Recommand extends Configured implements Tool {



    public static class IdIndexMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text>{

        public void map(LongWritable key, Text val, OutputCollector<Text, Text> output,Reporter reporter)throws IOException{
            String[] ids;
            String ln=val.toString();
            ids=ln.split("\t");
            output.collect(new Text(ids[0]),new Text(ids[1]));

    }
}
    public static class FtrMapper extends MapReduceBase implements Mapper<Text, Text, Text, Text>{
        public void map(Text key, Text val, OutputCollector<Text, Text>output, Reporter reporter) throws IOException{
            String[] str;

            String lne=val.toString();
        while(lne.contains("M1024")){
                str=lne.split(",");
            String[] str1=new String[str.length];
                for(int i=0;i<str.length;i++){
                                if(str[i]=="M1024"){   //hre need to give id which we need to split;
                                    continue;
                                        }
                                str1[i]=str[i];
                                output.collect(key,new Text(str1[i]));                  
//                          System.out.println("str1 out:"+str[i]); 
                                    }
                            }


        }
    }

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

    private final static IntWritable one=new IntWritable(1);
    private  Text word=new Text();
    public void map(Text key, Text val, OutputCollector<Text, IntWritable>output, Reporter reporter)throws IOException{
        String line = val.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, Iterable<IntWritable>values, OutputCollector<Text, IntWritable>output, Reporter reporter)throws IOException{
        int sum=0;
        for(IntWritable val:values){
            sum+=val.get();
                    }
        output.collect(key,new IntWritable(sum));
                    }
                }   

static int printUsage() {
    System.out.println("recommand  ");
    ToolRunner.printGenericCommandUsage(System.out);
    return -1;
}

public int run(String[] args) throws Exception {
    JobConf conf = new JobConf(getConf(), Recommand.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, IdIndexMapper.class, LongWritable.class, Text.class, Text.class, Text.class, true, mapAConf);

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

    JobConf mapCConf = new JobConf(false);
    ChainMapper.addMapper(conf, CntMapper.class, Text.class, Text.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 org.apache.hadoop.conf.Configuration(), Recommand(), args);
    System.exit(res);
}
}

谁能帮我解决这个问题？

Answer 1

确保以下内容以避免此错误

1. 这两个类都扩展了 Mapper 类。
2. 正在使用的 ChainMapper 类来自正确的 API，以适用于您的代码为准。

org.apache.hadoop.mapreduce.lib.chain.ChainMapper 或导入 org.apache.hadoop.mapred.lib.ChainMapper