求所有两两用户之间的共同好友
数据格式
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,K
以上是数据：
A:B,C,D,F,E,O
表示：B,C,D,E,F,O是A用户的好友。
public class SharedFriend {
/*
第一阶段的map函数主要完成以下任务
1.遍历原始文件中每行<所有朋友>信息
2.遍历“朋友”集合，以每个“朋友”为键，原来的“人”为值 即输出<朋友,人>
*/
static class SharedFriendMapper01 extends Mapper<LongWritable, Text, Text, Text>{
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] person_friends = line.split(":");
String person = person_friends[0];
String[] friends = person_friends[1].split(",");
for(String friend : friends){
context.write(new Text(friend), new Text(person));
}
}
}
/*
第一阶段的reduce函数主要完成以下任务
1.对所有传过来的<朋友，list(人)>进行拼接，输出<朋友,拥有这名朋友的所有人>
*/
static class SharedFriendReducer01 extends Reducer<Text, Text, Text, Text>{
@Override
protected void reduce(Text key, Iterable<Text> values,Context context)
throws IOException, InterruptedException {
StringBuffer sb = new StringBuffer();
for(Text friend : values){
sb.append(friend.toString()).append(",");
}
sb.deleteCharAt(sb.length()-1);
context.write(key, new Text(sb.toString()));
}
}
/*
第二阶段的map函数主要完成以下任务
1.将上一阶段reduce输出的<朋友,拥有这名朋友的所有人>信息中的 “拥有这名朋友的所有人”进行排序 ，以防出现B-C C-B这样的重复
2.将 “拥有这名朋友的所有人”进行两两配对，并将配对后的字符串当做键，“朋友”当做值输出，即输出<人-人，共同朋友>
*/
static class SharedFriendMapper02 extends Mapper<LongWritable, Text, Text, Text>{
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] friend_persons = line.split("\t");
String friend = friend_persons[0];
String[] persons = friend_persons[1].split(",");
Arrays.sort(persons); //排序
//两两配对
for(int i=0;i<persons.length-1;i++){
for(int j=i+1;j<persons.length;j++){
context.write(new Text(persons[i]+"-"+persons[j]+":"), new Text(friend));
}
}
}
}
/*
第二阶段的reduce函数主要完成以下任务
1.<人-人，list(共同朋友)> 中的“共同好友”进行拼接 最后输出<人-人，两人的所有共同好友>
*/
static class SharedFriendReducer02 extends Reducer<Text, Text, Text, Text>{
@Override
protected void reduce(Text key, Iterable<Text> values,Context context)
throws IOException, InterruptedException {
StringBuffer sb = new StringBuffer();
Set<String> set = new HashSet<String>();
for(Text friend : values){
if(!set.contains(friend.toString()))
set.add(friend.toString());
}
for(String friend : set){
sb.append(friend.toString()).append(",");
}
sb.deleteCharAt(sb.length()-1);
context.write(key, new Text(sb.toString()));
}
}
public static void main(String[] args)throws Exception {
Configuration conf = new Configuration();
//第一阶段
Job job1 = Job.getInstance(conf);
job1.setJarByClass(SharedFriend.class);
job1.setMapperClass(SharedFriendMapper01.class);
job1.setReducerClass(SharedFriendReducer01.class);
job1.setOutputKeyClass(Text.class);
job1.setOutputValueClass(Text.class);
FileInputFormat.setInputPaths(job1, new Path("H:/大数据/mapreduce/sharedfriend/input"));
FileOutputFormat.setOutputPath(job1, new Path("H:/大数据/mapreduce/sharedfriend/output"));
boolean res1 = job1.waitForCompletion(true);
//第二阶段
Job job2 = Job.getInstance(conf);
job2.setJarByClass(SharedFriend.class);
job2.setMapperClass(SharedFriendMapper02.class);
job2.setReducerClass(SharedFriendReducer02.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(Text.class);
FileInputFormat.setInputPaths(job2, new Path("H:/大数据/mapreduce/sharedfriend/output"));
FileOutputFormat.setOutputPath(job2, new Path("H:/大数据/mapreduce/sharedfriend/output01"));
boolean res2 = job2.waitForCompletion(true);
System.exit(res1?0:1);
}
}第一阶段输出结果
A F,I,O,K,G,D,C,H,B
B E,J,F,A
C B,E,K,A,H,G,F
D H,C,G,F,E,A,K,L
E A,B,L,G,M,F,D,H
F C,M,L,A,D,G
G M
H O
I O,C
J O
K O,B
L D,E
M E,F
O A,H,I,J,F
第二阶段输出结果
A-B C,E
A-C D,F
A-D E,F
A-E C,B,D
A-F E,O,C,D,B
A-G F,C,E,D
A-H D,O,C,E
A-I O
A-J B,O
A-K C,D
A-L D,E,F
A-M E,F
B-C A
B-D A,E
B-E C
B-F A,C,E
B-G E,C,A
B-H A,E,C
B-I A
B-K A,C
B-L E
B-M E
B-O K,A
C-D F,A
C-E D
C-F D,A
C-G D,F,A
C-H D,A
C-I A
C-K A,D
C-L D,F
C-M F
C-O I,A
D-E L
D-F A,E
D-G F,A,E
D-H A,E
D-I A
D-K A
D-L F,E
D-M F,E
D-O A
E-F C,D,M,B
E-G C,D
E-H C,D
E-J B
E-K D,C
E-L D
F-G C,E,D,A
F-H D,O,A,E,C
F-I A,O
F-J O,B
F-K D,C,A
F-L D,E
F-M E
F-O A
G-H E,C,D,A
G-I A
G-K D,A,C
G-L F,E,D
G-M E,F
G-O A
H-I A,O
H-J O
H-K C,D,A
H-L D,E
H-M E
H-O A
I-J O
I-K A
I-O A
K-L D
K-O A
L-M F,E