我正在尝试编写一个MapReduce应用程序,其中Mapper将一组值传递给Reducer,如下所示:
你好
世界
你好
你好
世界
你好
现在,首先对这些值进行分组和计数,然后进行一些进一步的处理。我写的代码是:
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
List<String> records = new ArrayList<String>();
/* Collects all the records from the mapper into the list. */
for (Text value : values) {
records.add(value.toString());
}
/* Groups the values. */
Map<String, Integer> groupedData = groupAndCount(records);
Set<String> groupKeys = groupedData.keySet();
/* Writes the grouped data. */
for (String groupKey : groupKeys) {
System.out.println(groupKey + ": " + groupedData.get(groupKey));
context.write(NullWritable.get(), new Text(groupKey + groupedData.get(groupKey)));
}
}
public Map<String, Integer> groupAndCount(List<String> records) {
Map<String, Integer> groupedData = new HashMap<String, Integer>();
String currentRecord = "";
Collections.sort(records);
for (String record : records) {
System.out.println(record);
if (!currentRecord.equals(record)) {
currentRecord = record;
groupedData.put(currentRecord, 1);
} else {
int currentCount = groupedData.get(currentRecord);
groupedData.put(currentRecord, ++currentCount);
}
}
return groupedData;
}
但是在输出中,我得到的计数为 1。sysout 语句的打印如下所示:
你好
世界
您好: 1
世界: 1
你好
您好: 1
你好
世界
您好: 1
世界: 1
你好
嗨:1
我不明白问题是什么,以及为什么Reducer没有立即接收所有记录并将其传递给groupAndCount
方法。
正如您在评论中指出的那样,如果每个值都有不同的相应键,则它们不会在相同的reduce调用中减少,并且您将获得当前看到的输出。
Hadoop reducer 的基本概念是,对于同一个键,值将被收集和减少 - 我建议你重新阅读一些 Hadoop 入门文档,尤其是字数统计示例,这似乎是你试图用代码实现的大致目标。