我有一个简单的mapreduce任务,就是从一个文本文件中读取一个字典,然后他们逐行处理另一个巨大的文件,并计算逆文档矩阵。输出应该如下所示:
word-id1 docX:tfX docY:tfY
word-id2 docX:tfX docY:tfY etc...
但是,减速机的输出仅在一条huuuge线中发出。我不明白为什么,因为它应该发出一个新的线为每一个 word-id
(这是减速器的关键)。
Map器生成正确的输出(一对 word-id
以及 doc-id:tf
在单独的行上)。我在没有减速器的情况下测试过。reducer应该只在一行中为每个键添加对应于同一个键的值。
你能看一下我的代码(特别是减速机和作业的配置)并告诉我为什么减速机只发出一个大的行而不是对应于指定键的多个行吗?我花了很多时间调试这个,不能使我的头周围。
public class Indexer extends Configured implements Tool {
/*
* Vocabulary: key = term, value = index
*/
private static Map<String, Integer> vocab = new HashMap<String, Integer>();
public static void main(String[] arguments) throws Exception {
System.exit(ToolRunner.run(new Indexer(), arguments));
}
public static class Comparator extends WritableComparator {
protected Comparator() {
super(Text.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
// Here we use exploit the implementation of compareTo(...) in
// Text.class.
return -a.compareTo(b);
}
}
public static class IndexerMapper extends
Mapper<Object, Text, IntWritable, Text> {
private Text result = new Text();
// load vocab from distributed cache
public void setup(Context context) throws IOException {
Configuration conf = context.getConfiguration();
FileSystem fs = FileSystem.get(conf);
URI[] cacheFiles = DistributedCache.getCacheFiles(conf);
Path getPath = new Path(cacheFiles[0].getPath());
BufferedReader bf = new BufferedReader(new InputStreamReader(
fs.open(getPath)));
String line = null;
while ((line = bf.readLine()) != null) {
StringTokenizer st = new StringTokenizer(line, " \t");
int index = Integer.parseInt(st.nextToken()); // first token is the line number - term id
String word = st.nextToken(); // second element is the term
// save vocab
vocab.put(word, index);
}
}
public void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
// init TF map
Map<String, Integer> mapTF = new HashMap<String, Integer>();
// parse input string
StringTokenizer st = new StringTokenizer(value.toString(), " \t");
// first element is doc index
int index = Integer.parseInt(st.nextToken());
// count term frequencies
String word;
while (st.hasMoreTokens()) {
word = st.nextToken();
// check if word is in the vocabulary
if (vocab.containsKey(word)) {
if (mapTF.containsKey(word)) {
int count = mapTF.get(word);
mapTF.put(word, count + 1);
} else {
mapTF.put(word, 1);
}
}
}
// compute TF-IDF
int wordIndex;
for (String term : mapTF.keySet()) {
int tf = mapTF.get(term);
if (vocab.containsKey(term)) {
wordIndex = vocab.get(term);
context.write(new IntWritable(wordIndex), new Text(index + ":" + tf));
}
}
}
}
public static class IndexerReducer extends Reducer<IntWritable, Text, IntWritable, Text>
{
@Override
public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException
{
StringBuilder sb = new StringBuilder(16000);
for (Text value : values)
{
sb.append(value.toString() + " ");
}
context.write(key, new Text(sb.toString()));
}
}
/**
* This is where the MapReduce job is configured and being launched.
*/
@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.setCombinerClass(IndexerReducer.class);
job.setReducerClass(IndexerReducer.class);
// Sort the output words in reversed order.
job.setSortComparatorClass(Comparator.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;
}
}
1条答案
按热度按时间ttcibm8c1#
尝试这些来调试您的问题-
将还原数设置为0,然后查看Map器的输出。
尝试使用默认的比较器,同样在你的比较器中你需要强制转换对象,否则它们不会产生正确的结果。