java - 实例化IdentityMapper得到了ClassException。如何使用IdentityMapper？

我有2个工作。第一个作业仅执行映射任务[1]，不执行归约任务。这强制将 map 输出保存在HDFS中。

该作业完成后，我将运行另一个具有标识映射器类[2]的作业，该类将读取前一个作业执行产生的映射输出，并产生相同的映射输出。我已经在[3]中将身份映射器设置为作业，但是在[4]中出现了错误。

我认为这个问题的原因是setMapperClass(Class<? extends Mapper> cls)来自不同类型的IdentityMapper [5]。

我如何使用IdentityMapper？

[1]第一份工作的 map 类

public static class MyMap extends 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, Context context
    ) throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }

    public void run(Context context) throws IOException, InterruptedException {
        try {
            while (context.nextKeyValue()) {
                map(context.getCurrentKey(), context.getCurrentValue(), context);
            }
        } finally {
            cleanup(context);
        }
    }
}

[2]身分类别:

public static class MyFullyIndentityMapper extends IdentityMapper<Text, IntWritable> {}

[3]我将标识类设置为映射器。

JobConf conf = new JobConf(MyWordCount.class);
conf.setJobName("wordcount");
conf.setClass("mapreduce.job.map.identity.class", MyFullyIndentityMapper.class, IdentityMapper.class);
Class<? extends IdentityMapper> identityClass = (Class<? extends IdentityMapper>) conf.getClass("mapreduce.job.map.identity.class", IdentityMapper.class);
job.setMapperClass(identityClass.asSubclass(Mapper.class));

[4]我得到的错误:

Output path: /output1
java.lang.ClassCastException: class org.apache.hadoop.mapred.examples.MyWordCount$MyFullyIndentityMapper
at java.lang.Class.asSubclass(Class.java:3404)

[5]身份映射器

public class IdentityMapper<K, V> extends MapReduceBase implements Mapper<K, V, K, V> {...}

最佳答案

我不明白为什么您在下面的代码片段中使事情变得复杂

JobConf conf = new JobConf(MyWordCount.class);
conf.setJobName("wordcount");
conf.setClass("mapreduce.job.map.identity.class", MyFullyIndentityMapper.class, IdentityMapper.class);
Class<? extends IdentityMapper> identityClass = (Class<? extends IdentityMapper>) conf.getClass("mapreduce.job.map.identity.class", IdentityMapper.class);
job.setMapperClass(identityClass.asSubclass(Mapper.class));

不要设置任何映射器类。

job.setMapperClass(identityClass.asSubclass(Mapper.class));

只需注释掉上面的行。
MapReduce框架将默认运行IdentityMapper。

以上是我对您问题的理解。我可能误解了您的问题。