MapReduce找共同好友

一、在网上找的题，就拿来做了

mapreduce笔试题:
找出有共同好友的users

A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G

O:A,H,I,J

最终结果：

================================================

A,B C,E

A,C D,F

A,D F,E

A,F B,C,D,E,O

B,E C

C,F A,D

D,E L

D,F A,E

D,L E,F

E,L D

F,M E

H,O A

二、代码部分

package com.wangs.jun;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FindFriends {
/**
* A：B,C,D
* 
* 
* B:A C:A D:A
* 
*/
public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
@Override
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
String[] split = value.toString().replaceAll(":", ",").split(",");
if (split.length > 0) {
Text v = new Text(split[0]);
Text k = new Text();
for (int i = 1; i < split.length; i++) {// 不能从0开始，因为0是值
k.set(split[i]);
context.write(k, v);
}
}
}
}

/**
* B:A , B:M, B，F
* 
* A,M:B A,F:B M,F:B A,B:1
* 
*/
public static class R1 extends Reducer<Text, Text, Text, Text> {

@Override
protected void reduce(Text key, Iterable<Text> values,
Reducer<Text, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
List<String> f = new ArrayList<>();// A,M,F
for (Text s : values) {
f.add(s.toString());
}
String[] ss = f.toArray(new String[0]);
for (int i = 0; i < ss.length; i++) {// M
String s1 = ss[i];
for (int j = 0; j < ss.length; j++) {// F
String s2 = ss[j];
if (!s1.equals(s2) && s1.compareTo(s2) < 0) {
context.write(new Text(s1 + "," + s2), key);
}
}

String ff = s1 + "," + key.toString();// M,B
if (s1.compareTo(key.toString()) > 0) {// B,M:1
ff = key.toString() + "," + s1;
}
context.write(new Text(ff), new Text("1"));
}
}
}

/**
* A,M:B A,M:C A,M:1
* 
* 
* @author liuxing
* 
*/
public static class M2 extends Mapper<LongWritable, Text, Text, Text> {
@Override
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
String[] split = value.toString().split("\t");
context.write(new Text(split[0]), new Text(split[1]));
}
}

// A,M:B,C,1,1
public static class R2 extends Reducer<Text, Text, Text, Text> {

@Override
protected void reduce(Text key, Iterable<Text> values,
Reducer<Text, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
int count = 0;
List<String> s = new ArrayList<>();
for (Text t : values) {
if ("1".equals(t.toString()))
count += 1;
else {
s.add(t.toString());
}
}
if (count == 2 && s.size() > 0) {
StringBuffer sb = new StringBuffer();
for (String ss : s) {
sb.append(",").append(ss);
}
context.write(key, new Text(sb.substring(1).toString()));
}
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(FindFriends.class);

job.setMapperClass(M1.class);
job.setReducerClass(R1.class);

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

FileInputFormat.setInputPaths(job, new Path(args[0]));
Path path = new Path(args[1]);

FileOutputFormat.setOutputPath(job, path);
boolean b = job.waitForCompletion(true);

if (b) {
Job job2 = Job.getInstance(conf);
job2.setJarByClass(FindFriends.class);
job2.setMapperClass(M2.class);
job2.setReducerClass(R2.class);

job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(Text.class);

FileInputFormat.setInputPaths(job2, new Path(args[1]));
Path path2 = new Path(args[2]);

FileOutputFormat.setOutputPath(job2, path2);

job2.waitForCompletion(true);
}
}
}