我有一个简单的 map reduce 作业,我正在其中构建反向索引。
我的映射器工作正常(我检查过)并输出 key 对 word 和 docID:TFIDF 值:
映射器(仅显示输出):
context.write(new IntWritable(wordIndex), new Text(index + ":" + tfidf));
reducer 唯一的工作就是合并这些值。这是我的实现:
public static class IndexerReducer extends Reducer<Text, IntWritable, IntWritable, Text>
{
public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
StringBuilder sb = new StringBuilder();
for (Text value : values)
{
sb.append(value.toString() + " ");
}
context.write(key, new Text(sb.toString()));
}
}
但是,它没有组合任何东西,输出看起来与映射器基本相同。尽管 reducer 应该将它们组合在一起,但输出中有些行具有相同的键 - 基本上输出文件中的所有键在使用 reducer 时都应该是唯一的,对吧?
这是我的 reducer 输出的示例(请注意,这是简化的示例):
1 15:2.1
1 13:4.3
2 9:9.3
2 43:7.9
etc
我期望这样:
1 15:2.1 13:4.3
2 9:9.3 43:7.9
为了完整起见,我包括了运行方法:
@Override
public int run(String[] arguments) throws Exception {
ArgumentParser parser = new ArgumentParser("TextPreprocessor");
parser.addArgument("input", true, true, "specify input directory");
parser.addArgument("output", true, true, "specify output directory");
parser.parseAndCheck(arguments);
Path inputPath = new Path(parser.getString("input"));
Path outputDir = new Path(parser.getString("output"));
// Create configuration.
Configuration conf = getConf();
// add distributed file with vocabulary
DistributedCache
.addCacheFile(new URI("/user/myslima3/vocab.txt"), conf);
// Create job.
Job job = new Job(conf, "WordCount");
job.setJarByClass(IndexerMapper.class);
// Setup MapReduce.
job.setMapperClass(IndexerMapper.class);
job.setReducerClass(IndexerReducer.class);
// Sort the output words in reversed order.
job.setSortComparatorClass(WordCountComparator.class);
job.setNumReduceTasks(1);
// Specify (key, value).
job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(Text.class);
// Input.
FileInputFormat.addInputPath(job, inputPath);
job.setInputFormatClass(TextInputFormat.class);
// Output.
FileOutputFormat.setOutputPath(job, outputDir);
job.setOutputFormatClass(TextOutputFormat.class);
FileSystem hdfs = FileSystem.get(conf);
// Delete output directory (if exists).
if (hdfs.exists(outputDir))
hdfs.delete(outputDir, true);
// Execute the job.
return job.waitForCompletion(true) ? 0 : 1;
}
如果有任何关于正在发生的事情的提示,我会很高兴。我是 map 减少的新手。感谢您提供任何调试提示!
最佳答案
始终使用 @Override
注释。
你定义了
public static class IndexerReducer extends Reducer<Text, IntWritable, IntWritable, Text>
那么你的 reduce 方法必须看起来像那样
@Override
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException
关于java - MapReduce - reducer 不组合键,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29394443/