mapreduce-找出共同好友
1、原始資料
A:B,C,D,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
2、解題思路：
第一步：
map
讀一行 A:B,C,D,E,O
輸出 <B,A><C,A><D,A><E,A><O,A>
再讀一行 B:A,C,E,K
輸出 <A,B><C,B><E,B><K,B>reduce操作：
拿到的資料比如說：<C,A><C,B><C,E><C,F><C,G>.......
輸出
<A-B,C><A-E,C><A-F,C><A-G,C><B-E,C><B-F,C>.......第二步：
map:
讀入一行： <A-B,C>
直接輸出： <A-B,C>reduce操作：
讀入資料：<A-B,C><A-B,D><A-B,F>......
輸出：A-B C,F,G
3、程式碼實現（分兩步）
第一步：
package cn.itacst.mr.ShareParents; import java.io.IOException; 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 ShareParentsStepOne { static class ShareParentsStepOneMapper extends Mapper<LongWritable, Text,Text, Text>{ protected void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException { //A:B,C,D,E,O 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)); } }; } static class ShareParentsStepOnReducer extends Reducer<Text, Text, Text,Text>{ @Override protected void reduce(Text friend, Iterable<Text> persons,Context context) throws IOException, InterruptedException { StringBuffer sb = new StringBuffer(); for(Text person:persons){ sb.append(person).append(","); } context.write(friend, new Text(sb.toString())); } } public static void main(String[] args) throws Exception { if (args.length == 0) { args = new String[2]; args[0] = "hdfs://192.168.40.10:9000/qqShareFriends/input/"; args[1] = "hdfs://192.168.40.10:9000/qqShareFriends/output1/"; } Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(ShareParentsStepOne.class); job.setMapperClass(ShareParentsStepOneMapper.class); job.setReducerClass(ShareParentsStepOnReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.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); } }
第二步：
package cn.itacst.mr.ShareParents;
import java.io.IOException;
import java.util.Arrays;
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.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import cn.itacst.mr.ShareParents.ShareParentsStepOne.ShareParentsStepOnReducer;
import cn.itacst.mr.ShareParents.ShareParentsStepOne.ShareParentsStepOneMapper;
public class ShareParentsStepTwo {
static class ShareParentsStepTwoMapper extends Mapper<LongWritable, Text,Text, Text>{
protected void map(LongWritable key, Text value, Context context)
throws java.io.IOException ,InterruptedException {
//A C,O,I,B,F,G,D,H,K,
//(友，人1，人2，人3...)連線成為人人對
String line = value.toString();
//切分
String[] friend_person = line.split("\t");
String friend = friend_person[0];
String[] person = friend_person[1].split(",");
Arrays.sort(person);
for(int i=0;i<person.length-2;i++){
for(int j=i+1;j<person.length-1;j++){
context.write(new Text(person[i]+"-"+person[j]), new Text(friend));
}
}
};
}
static class ShareParentsStepTwoReducer extends Reducer<Text, Text, Text,Text>{
@Override
protected void reduce(Text person_person, Iterable<Text> friends,Context context)
throws IOException, InterruptedException {
StringBuffer sb = new StringBuffer();
for(Text friend:friends){
sb.append(friend).append(" ");
}
context.write(person_person, new Text(sb.toString()));
}
}
public static void main(String[] args) throws Exception {
if (args.length == 0) {
args = new String[2];
args[0] = "hdfs://192.168.40.10:9000/qqShareFriends/output1";
args[1] = "hdfs://192.168.40.10:9000/qqShareFriends/output2";
}
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(ShareParentsStepTwo.class);
job.setMapperClass(ShareParentsStepTwoMapper.class);
job.setReducerClass(ShareParentsStepTwoReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.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);
}
}
4、結果
第一步結果：
A C,O,I,B,F,G,D,H,K,
B J,F,E,A,
C H,B,G,F,K,A,E,
D C,L,K,A,H,G,F,E,
E H,F,L,M,B,A,D,G,
F C,L,D,M,G,
G M,
H O,
I O,C,
J O,
K B,
L D,E,
M E,F,
O A,H,I,J,F,
第二步結果：
A-B E C
A-C D
A-D E
A-E D B C
A-F D O C E B
A-G E C D
A-H E O C D
A-I O
A-K D
A-L E
B-C A
B-D E A
B-E C
B-F E A C
B-G C E A
B-H E C A
B-I A
B-K A
B-L E
C-D F A
C-E D
C-F D A
C-G F A D
C-H A D
C-I A
C-K D A
C-L F
D-F E A
D-G A E F
D-H A E
D-I A
D-K A
D-L F E
E-F C D B
E-G D C
E-H D C
E-K D
F-G C E D A
F-H C A D E O
F-I A O
F-K D A
F-L E
G-H D E C A
G-I A
G-K A D
G-L F E
H-I A O
H-K A D
H-L E
I-K A