java - MapReduce 程序映射任务超时

Question

我遇到了这个奇怪的错误。我编写了一个 wordCount 程序来计算一个单词在文件中重复的次数。

所以当我在 hadoop 上运行 MR 代码时，代码卡在“Map 100%, Reduce 0%”。基本模式是第一个 map task 在 600 秒后超时，然后再次超时，任务自行终止。

我检查了 Job Tracker，任务卡住了，因为 Map 任务没有完成以等待 reduce 任务开始。

我已经尝试修复它 2 天，在此期间我删除了原始虚拟 Ubuntu Cloudera 并重新安装它 - 所以我们可以确定这不是配置问题。

感谢任何帮助。

以下是3个代码文件。

WordCount.java

public class WordCount extends Configured implements Tool {

@Override
public int run(String[] args) throws Exception {

    Configuration conf =  super.getConf();

    Job job=new Job(conf, "Word Count Job");
    job.setJarByClass(WordCount.class);

    job.setMapperClass(WordMapper.class);
    job.setReducerClass(WordReducer.class);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FloatWritable.class);

    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);

    FileInputFormat.setInputPaths(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    job.waitForCompletion(Boolean.TRUE);

    return 0;
}

public static void main(String[] args) {

    //Display error message in case insufficient arguments supplied
    if(args.length<2){
        System.out.println("usage: WordCount <Input-Path> <Output-Path>");
    }

    Configuration conf=new Configuration(Boolean.TRUE);

    int i;
    try {
        //Run the overridden 'run' method code
        i = ToolRunner.run(conf, new WordCount(), args);

        //Print usage stats to out
        //ToolRunner.printGenericCommandUsage(System.out);

        //exit if job cannot start
        System.exit(i);

    } catch (Exception e) {

        e.printStackTrace();

        System.exit(-1);
    }
}
}

WordMapper.java

public class WordMapper extends Mapper<LongWritable, Text, Text, FloatWritable> {

@Override
protected void map(LongWritable key, 
        Text value,
        Mapper<LongWritable, Text, Text, FloatWritable>.Context context)
        throws IOException, InterruptedException {


    if(!value.toString().trim().isEmpty()){

        StringTokenizer valTokens = new StringTokenizer(value.toString()); 

        while(valTokens.hasMoreTokens()){
            context.write(new Text(valTokens.nextToken()), new FloatWritable(Float.parseFloat("1.00")));
        }
    }   
}
}

WordReducer.java

public class WordReducer extends Reducer<Text, FloatWritable, Text, FloatWritable> {

@Override
protected void reduce(Text key, Iterable<FloatWritable> values,
        Reducer<Text, FloatWritable, Text, FloatWritable>.Context context)
        throws IOException, InterruptedException {

    Iterator<FloatWritable> valsIter = values.iterator();
    int i = 0;

    while(valsIter.hasNext()) 
        i++;

    context.write(key, new FloatWritable((float)i));
}
}

Answer 1

您的问题出在这行代码中:

    while(valsIter.hasNext()) 
    i++;

valsIter.hasNext 检查迭代器中是否有下一个元素但不移动指针的位置。因此检查总是返回 true。除非你调用 valsIter.next()。