我正在使用 MapReduce 在 Hadoop 中开发一个非常简单的图形分析工具。我有一个如下图（每行代表和边缘 - 事实上，这是一个三角形图）：
1 3
3 1
3 2
2 3
现在，我想使用 MapReduce 来计算该图中的三角形（显然是一个）。它仍在进行中，在第一阶段，我尝试获取每个顶点的所有邻居列表。
我的主要课程如下所示：
public class TriangleCount {
public static void main( String[] args ) throws Exception {
// remove the old output directory
FileSystem fs = FileSystem.get(new Configuration());
fs.delete(new Path("output/"), true);
JobConf firstPhaseJob = new JobConf(FirstPhase.class);
firstPhaseJob.setOutputKeyClass(IntWritable.class);
firstPhaseJob.setOutputValueClass(IntWritable.class);
firstPhaseJob.setMapperClass(FirstPhase.Map.class);
firstPhaseJob.setCombinerClass(FirstPhase.Reduce.class);
firstPhaseJob.setReducerClass(FirstPhase.Reduce.class);
FileInputFormat.setInputPaths(firstPhaseJob, new Path("input/"));
FileOutputFormat.setOutputPath(firstPhaseJob, new Path("output/"));
JobClient.runJob(firstPhaseJob);
}
}
我的 Mapper 和 Reducer 实现看起来像这样，它们都很简单：
public class FirstPhase {
public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, IntWritable, IntWritable> {
@Override
public void map(LongWritable longWritable, Text graphLine, OutputCollector<IntWritable, IntWritable> outputCollector, Reporter reporter) throws IOException {
StringTokenizer tokenizer = new StringTokenizer(graphLine.toString());
int n1 = Integer.parseInt(tokenizer.nextToken());
int n2 = Integer.parseInt(tokenizer.nextToken());
if(n1 > n2) {
System.out.println("emitting (" + new IntWritable(n1) + ", " + new IntWritable(n2) + ")");
outputCollector.collect(new IntWritable(n1), new IntWritable(n2));
}
}
}
public static class Reduce extends MapReduceBase implements Reducer<IntWritable, IntWritable, IntWritable, Text> {
@Override
public void reduce(IntWritable key, Iterator<IntWritable> iterator, OutputCollector<IntWritable, Text> outputCollector, Reporter reporter) throws IOException {
List<IntWritable> nNodes = new ArrayList<>();
while(iterator.hasNext()) {
nNodes.add(iterator.next());
}
System.out.println("key: " + key + ", list: " + nNodes);
// create pairs and emit these
for(IntWritable n1 : nNodes) {
for(IntWritable n2 : nNodes) {
outputCollector.collect(key, new Text(n1.toString() + " " + n2.toString()));
}
}
}
}
}
我在程序中添加了一些日志记录。在映射阶段，我打印出我正在发射的对。在 reduce 阶段，我打印 reduce 的输入。我得到以下输出：
emitting (3, 1)
emitting (3, 2)
key: 3, list: [1, 1]
reduce 函数的输入不是我所期望的。我希望它是 [1, 2] 而不是 [1, 1]。我相信 Hadoop 会自动组合我从 map 阶段的输出中发出的所有对，但我在这里遗漏了什么吗？任何帮助或解释将不胜感激。