MapReduce—案例（五）求两两共同好友

题目：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连接起来代码：package practice1;

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.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.jobcontrol.JobControl;
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.output.FileOutputFormat;

public class Practice4 {

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();
// conf.set("fs.defaultFS", "hdfs://potter2:9000");//使用配置文件
// System.setProperty("HADOOP_USER_NAME", "potter");//使用集群
////////////第一步/////////////////////////
FileSystem fs = FileSystem.get(conf);
Job job = Job.getInstance();
job.setJarByClass(Practice4.class);
job.setMapperClass(Practice4Mapper.class);
job.setReducerClass(Practice4Reducer.class);

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

Path input = new Path("D:\\practice\\input4\\work4.txt");
Path output = new Path("D:\\practice\\input4\\output1");

FileInputFormat.setInputPaths(job, input);
FileOutputFormat.setOutputPath(job, output);

if (fs.exists(output)) {
fs.delete(output,true);
}
/////////////第二步///////////////////////
FileSystem fs2 = FileSystem.get(conf);

Job job2 = Job.getInstance();
job2.setJarByClass(Practice4.class);
job2.setMapperClass(Practice4_Mapper.class);
job2.setReducerClass(Practice4_Reducer.class);
job2.setMapOutputKeyClass(Text.class);
job2.setMapOutputValueClass(Text.class);
job2.setMapOutputKeyClass(Text.class);
job2.setOutputValueClass(Text.class);

Path input2 = new Path("D:\\practice\\input4\\output1");
Path output2 = new Path("D:\\practice\\input4\\output2");
FileInputFormat.setInputPaths(job2, input2);
FileOutputFormat.setOutputPath(job2, output2);

if (fs2.exists(output2)) {
fs2.delete(output2,true);
}
//采用jobcontrol进行两个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("两两用户之间的共同好友");

jc.addJob(ajob);
jc.addJob(bjob);

Thread thread = new Thread(jc);
thread.start();
while (!jc.allFinished()) {

Thread.sleep(500);

}
jc.stop();
// boolean isdone = job.waitForCompletion(true);
// System.exit(isdone ? 0 :1);

}
//第一个MapReduce
public static class Practice4Mapper extends Mapper<LongWritable, Text, Text, Text>{
/**
* A:B,C,D,F,E,O
* B:A,C,E,K
*/
Text text1 = new Text();
Text text2 = new Text();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
//A:B,C,D,F,E,O
String[] split1 = value.toString().split(":");
//以用户为value值，以好有为key值，求某用户为共同好有的人
String aa = split1[0];
text1.set(aa);

String[] split2 = split1[1].split(",");

for (int i = 0; i < split2.length; i++) {
String ll = split2[i];
text2.set(ll);
context.write(text2, text1);
}
}

}

public static class Practice4Reducer extends Reducer<Text, Text,Text, Text>{
Text text3 = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
//将用户共同的好有组合起来
StringBuilder sb = new StringBuilder();
for(Text text : values){
sb.append(text.toString()).append(",");
}
String xx = sb.substring(0,sb.length()-1);
text3.set(xx);
context.write(key, text3);
}
}

//第二个MapReduce
public static class Practice4_Mapper extends Mapper<LongWritable, Text, Text, Text>{
//A F,I,O,K,G,D,C,H,B
//输出的意思是：F,I,O,K,G,D,C,H,B的共同好有是A
//第二步需要求取两两好有，以A为value，互为好友对为key
Text text1 = new Text();
Text text2 = new Text();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
String[] split1 = value.toString().split("\t");
//A为value
String aa = split1[0];
text1.set(aa);

String[] split2 = split1[1].split(",");
//求两两好友对，为了防止重复，需要对好友进行排序
Arrays.sort(split2);
//使用双重循环，寻找A-B等好友对组合
for (int i = 0; i < split2.length-1; i++) {
for (int j = i + 1; j < split2.length; j++) {
String dd = split2[i]+"-"+split2[j];
text2.set(dd);
context.write(text2, text1);
}
}
}
}

public static class Practice4_Reducer extends Reducer<Text, Text, Text, Text>{
Text text3 = 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 cc = sb.substring(0,sb.length()-1);
text3.set(cc);
context.write(key, text3);
}
}
}