因此，我有一个巨大的访问日志文件，我试图找到的路径上的服务器是击中最多。查找路径被击中的次数是一个传统的字数计算问题。
但是，由于输出值在mr作业中没有排序（只对键进行排序），因此我正在执行另一个mr作业，Map器将上一个作业的输出作为输入，我使用inversemapper.java反转键和值，并使用identity reducer（reducer.java），因为不需要聚合，我只需要对键进行排序（即。，第一份工作的价值观）。这是我的密码：

package edu.pitt.cloud.CloudProject;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable.DecreasingComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.map.InverseMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class AccessLogMostHitPath {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        String configPath = "/usr/local/hadoop-2.7.3/etc/hadoop/";

        Path inputPath = new Path(args[0]);
        Path outputPath = new Path(args[1]);
        Path finalOutputPath = new Path(args[2]);

        Configuration config = new Configuration(true);

        config.addResource(new Path(configPath+"hdfs-site.xml"));
        config.addResource(new Path(configPath+"core-site.xml"));
        config.addResource(new Path(configPath+"yarn-site.xml"));
        config.addResource(new Path(configPath+"mapred-site.xml"));

        Job job = Job.getInstance(config, "AccessLogMostHitPath");
        job.setJarByClass(AccessLogMostHitPath.class);

        job.setMapperClass(AccessLogMostHitPathMapper.class);
        job.setReducerClass(AccessLogMostHitPathReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        config.set("mapreduce.job.running.map.limit", "2");

        FileInputFormat.addInputPath(job, inputPath);
        job.setInputFormatClass(TextInputFormat.class);

        FileOutputFormat.setOutputPath(job, outputPath);
        job.setOutputFormatClass(TextOutputFormat.class);

        job.setNumReduceTasks(1);   

        System.out.println("Starting Job Execution ::: AccessLogMostHitPath");
        int code = job.waitForCompletion(true) ? 0 : 1;
        System.out.println("Job Execution Finished ::: AccessLogMostHitPath");

        if(code != 0){
            System.out.println("First Job Failed");
            System.exit(code);
        }

            FileSystem hdfs = FileSystem.get(config);
            Path successPath = new Path(outputPath+"/_SUCCESS");
            if (hdfs.exists(successPath))
              hdfs.delete(successPath, true);

        Job job2 = Job.getInstance(config, "AccessLogMostHitPathSort");
        job2.setJarByClass(AccessLogMostHitPath.class);

        job2.setMapperClass(InverseMapper.class);
        job2.setReducerClass(Reducer.class);
        //config.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", "\\t");

        KeyValueTextInputFormat.addInputPath(job2, outputPath);
        job2.setInputFormatClass(KeyValueTextInputFormat.class);

        FileOutputFormat.setOutputPath(job2, finalOutputPath);
        job2.setOutputFormatClass(TextOutputFormat.class);
        job2.setNumReduceTasks(1);

        job2.setMapOutputKeyClass(IntWritable.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setSortComparatorClass(DecreasingComparator.class);

        job2.setOutputKeyClass(IntWritable.class);
        job2.setOutputValueClass(Text.class);
        config.set("mapreduce.job.running.map.limit", "2");

        System.out.println("Starting Job Execution ::: AccessLogMostHitPathSort");
        int code2 = job2.waitForCompletion(true) ? 0 : 1;
        System.out.println("Job Execution Finished ::: AccessLogMostHitPathSort");
        System.exit(code2);

    }
}

执行此操作时，出现以下异常：

Error: java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.IntWritable, received org.apache.hadoop.io.Text
        at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1072)
        at org.apache.hadoop.mapred.MapTask$NewOutputCollector.write(MapTask.java:715)
        at org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write(TaskInputOutputContextImpl.java:89)
        at org.apache.hadoop.mapreduce.lib.map.WrappedMapper$Context.write(WrappedMapper.java:112)
        at org.apache.hadoop.mapreduce.lib.map.InverseMapper.map(InverseMapper.java:36)
        at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:146)
        at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:787)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:164)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1698)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)

哪里出了问题？。我可以看到某个地方有一个不匹配的键或值类型，但我已经交叉检查了一切。请帮忙。