Mapreduce实例---共同好友

public class CommonFriendsStepOne {
public static class CommonFriendsStepOneMapper extends Mapper<LongWritable, Text, Text, Text>{
private Text k = new Text();
private Text v = new Text();

// A:B,C,D,F,E,O       B--> A ; C-->A ; D-->A ; F-->A ....
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
String line = value.toString();
String[] split = line.split(":");
String[] friends = split[1].split(",");

v.set(split[0]);
for(String f : friends){
k.set(f);
context.write(k, v);
}
}
}

public static class CommonFriendsStepOneReducer extends Reducer<Text, Text, Text, Text>{
/*
* 输入的数据 <B A> <B E> <B F> <B J>
* 输出的结果 <B A,E,F,J>
*/
private Text v=new Text();
@Override
protected void reduce(Text friend, Iterable<Text> persons,Context context)throws IOException, InterruptedException {
StringBuilder sb=new StringBuilder();
for(Text p : persons){
sb.append(p).append(",");
}
v.set(sb.toString());
context.write(friend, v);
}
}

public static void main(String[] args) throws Exception, IOException {
Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(CommonFriendsStepOne.class);

job.setMapperClass(CommonFriendsStepOneMapper.class);
job.setReducerClass(CommonFriendsStepOneReducer.class);

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

FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

boolean res = job.waitForCompletion(true);

System.exit(res ? 0 : 1);
}
}

public class CommonFriendsStepTwo {
static class CommonFriendsStepTwoMapper extends Mapper<LongWritable, Text, Text, Text>{
//A I,K,C,B,G,F,H,O,D,       B A,F,J,E,

private Text k=new Text();
private Text v=new Text();
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
String line = value.toString();
String[] split = line.split("\t");
String[] persons = split[1].split(",");

// 对用户进行排序，以免出现GF、FG被视为不同的组合
Arrays.sort(persons);
v.set(split[0]);

// 对整个persons数组做两两组合拼接
for(int i=0;i<persons.length-1;i++){
for(int j=i+1;j<persons.length;j++){
// 输出 <I-J A>
k.set(persons[i] + "-" + persons[j]);
context.write(k, v);
}
}
}
}

static class CommonFriendsStepTwoReducer extends Reducer<Text, Text, Text, Text>{
// <A-E,B> <A-E,C>.....
private Text v = new Text();

@Override
protected void reduce(Text pair, Iterable<Text> friends,Context context)throws IOException, InterruptedException {
StringBuilder sb=new StringBuilder();
for(Text f : friends){
sb.append(f).append(" ");
}
// <A-E, B C ..>
v.set(sb.toString());
context.write(pair, v);
}
}
public static void main(String[] args) throws Exception, IOException {
Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(CommonFriendsStepTwo.class);

job.setMapperClass(CommonFriendsStepTwoMapper.class);
job.setReducerClass(CommonFriendsStepTwoReducer.class);

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

FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

boolean res = job.waitForCompletion(true);

System.exit(res ? 0 : 1);
}
}