MapReduce案例4——求两两共同好友

题目如下：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用户的好友。

1、求所有两两用户之间的共同好友题意理解：求两两的共同好友，例如c是A的好友，C是B的好友，所有AB的共同好友有C。题目陷阱：I的好友包括A，但是A的好友没有I;
解题思路：采用逆向思维，分两步走：要求两两的共同好友，可以先求出以A为共同好友的人的组合，第二步：将组合好的人，两两组合，求出所有的好友。采用组件jobcontrol将两个MapReduce连接起来
代码：/**
* @author: lpj
* @date: 2018年3月16日 下午7:16:47
* @Description:
*/
package lpj.reduceWork;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
*
*/
public class SameFriendsMR {

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
// conf.addResource("hdfs-site.xml");//使用配置文件
// System.setProperty("HADOOP_USER_NAME", "hadoop");//使用集群
///////////////////第一步////////////////////////
FileSystem fs = FileSystem.get(conf);//默认使用本地

Job job = Job.getInstance(conf);
job.setJarByClass(SameFriendsMR.class);
job.setMapperClass(SameFriendsMR_Mapper.class);
job.setReducerClass(SameFriendsMR_Reducer.class);

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

Path inputPath = new Path("d:/a/homework4.txt");
Path outputPath = new Path("d:/a/homework4");
if (fs.exists(inputPath)) {
fs.delete(outputPath, true);
}

FileInputFormat.setInputPaths(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
////////////////////第二步//////////////////////////////////
FileSystem fs2 = FileSystem.get(conf);//默认使用本地

Job job2 = Job.getInstance(conf);
job2.setJarByClass(SameFriendsMR.class);
job2.setMapperClass(SameFriends2MR_Mapper.class);
job2.setReducerClass(SameFriends2MR_Reducer.class);

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

Path inputPath2 = new Path("d:/a/homework4");
Path outputPath2 = new Path("d:/a/homework4_1");
if (fs2.exists(inputPath2)) {
fs2.delete(outputPath2, true);
}

FileInputFormat.setInputPaths(job2, inputPath2);
FileOutputFormat.setOutputPath(job2, outputPath2);

//采用jobcontrol进行2个MapReduce串行
ControlledJob aJob = new ControlledJob(job.getConfiguration());
ControlledJob bJob = new ControlledJob(job2.getConfiguration());

aJob.setJob(job);
bJob.setJob(job2);
bJob.addDependingJob(aJob);//指定依赖关系

JobControl jc = new JobControl("jcF");
jc.addJob(aJob);
jc.addJob(bJob);

Thread thread = new Thread(jc);
thread.start();
while(!jc.allFinished()){
thread.sleep(1000);
}
jc.stop();
}
///////////////////////////////第一步/////////////////////////////////////////
public static class SameFriendsMR_Mapper extends Mapper<LongWritable, Text, Text, Text>{
Text kout = new Text();
Text valueout = new Text();
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
//A:B,C,D,F,E,O
String [] reads = value.toString().trim().split(":");
//以用户为value值，好友为key值，求取以某用户为共同好友的人
String vv = reads[0];
valueout.set(vv);
String [] friends = reads[1].split(",");
for(int i = 0; i < friends.length; i++){
String kk = friends[i];
kout.set(kk);
context.write(kout, valueout);
}
}
}
public static class SameFriendsMR_Reducer extends Reducer<Text, Text, Text, Text>{
Text kout = new Text();
Text valueout = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
//将用户共同好友的人组合起来
StringBuilder sb = new StringBuilder();
for(Text text : values){
sb.append(text.toString()).append(",");
}
String vv = sb.substring(0, sb.length() - 1);
valueout.set(vv);
context.write(key, valueout);
}

}
//////////////////////////////////////第二步///////////////////////////////////////////////////////////
public static class SameFriends2MR_Mapper extends Mapper<LongWritable, Text, Text, Text>{
Text kout = new Text();
Text valueout = new Text();
@Override
protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {
//A F,I,O,K,G,D,C,H,B ：意思是F,I,O,K,G,D,C,H,B的共同好友为A,第二步执行求取两两好友,A为value，两两对为key
String [] reads = value.toString().trim().split("\t");
//A为value
String vv = reads[0];
valueout.set(vv);
//求两两好友对，为了防止重复，需要对好友进行排序
String [] friends = reads[1].split(",");
Arrays.sort(friends);
//使用两重循环，寻找A-B等好友对组合
for(int i = 0; i < friends.length - 1; i++){
for(int j = i + 1; j < friends.length; j++ ){
String kk = friends[i] + "-" + friends[j];
kout.set(kk);
context.write(kout, valueout);
}
}
}
}
public static class SameFriends2MR_Reducer extends Reducer<Text, Text, Text, Text>{
Text kout = new Text();
Text valueout = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
//好友对组合起来
StringBuilder sb = new StringBuilder();
for(Text text : values){
sb.append(text.toString()).append(",");
}
String vv = sb.substring(0, sb.length() - 1);
valueout.set(vv);
context.write(key, valueout);
}

}
}第一步运行结果：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第二步运行结果：共74行A-B E,C
A-C D,F
A-D E,F
A-E B,C,D
A-F C,E,O,D,B
A-G E,F,C,D
A-H C,D,E,O
A-I O
A-J O,B
A-K C,D
A-L F,D,E
A-M F,E
B-C A
B-D A,E
B-E C
B-F C,A,E
B-G E,C,A
B-H E,C,A
B-I A
B-K A,C
B-L E
B-M E
B-O A,K
C-D A,F
C-E D
C-F A,D
C-G A,D,F
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 E,A,F
D-H A,E
D-I A
D-K A
D-L E,F
D-M F,E
D-O A
E-F D,M,C,B
E-G C,D
E-H C,D
E-J B
E-K C,D
E-L D
F-G D,C,A,E
F-H A,D,O,E,C
F-I O,A
F-J B,O
F-K D,C,A
F-L E,D
F-M E
F-O A
G-H D,C,E,A
G-I A
G-K D,A,C
G-L D,F,E
G-M E,F
G-O A
H-I O,A
H-J O
H-K A,C,D
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 E,F