最近有了解到一些数据开发的面试提会要求开发进行手动编写map-reduce；记录一下使用map-reduce进行数据处理的过程及逻辑；
现在版本的map-reduce过程基本都依赖新的jar包，之前的mapred相关的类后面不被推荐使用，使用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
代码及注释：
思路
1、先map遍历列表，获取 用户->好友 的映射关系 输出 好友->用户 一对一关系
2、reduce循环遍历，获取 好友（共同的好友）->用户list
3、再进行一次map，对用户列表遍历，获取两两用户的key 和 好友value 例如 A-B (key) : E (value) 输出到reduce
4、reduce进行迭代组合根据两两用户(A-B)合并 好友(E)的列表.
package hadoop;
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;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
public class CommonFriends {
/*
* @param LongWritable map根据输入的偏移量作为key
* @param Text 第一个是 文件的记录
* @param Text 第二个是 map输出的key类型
* @param Text 第三个是 map输出的value类型
* */
public static class commonFriendMap01 extends Mapper<LongWritable, Text, Text, Text>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] per_friend = line.split(":");
String per = per_friend[0];
String[] friends = per_friend[1].split(",");
for(String f:friends){
context.write(new Text(f),new Text(per));
}
}
}
/*
* @param Text 第一个是 reduce输入的key类型
* @param Text 第二个是 reduce输入的value类型
* @param Text 第三个是 reduce输出的key类型
* @param Text 第四个是 reduce输出的value类型
* */
public static class commonFriendReduce01 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<>();
//使用set进行去重操作
for(Text f:values){
if(!set.contains(f.toString())){
set.add(f.toString());
}
}
for(String f:set){
sb.append(f).append(",");
}
sb.deleteCharAt(sb.length()-1);
context.write(key,new Text(sb.toString()));
}
}
/*
* @param LongWritable map根据输入的偏移量作为key
* @param Text 第一个是 文件的记录
* @param Text 第二个是 map输出的key类型
* @param Text 第三个是 map输出的value类型
* */
public static class commonFriendMap02 extends Mapper<LongWritable,Text,Text,Text>{
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String [] per_fr = value.toString().split("\t");
String friend = per_fr[0];
String[] per = per_fr[1].split(",");
//如果存在乱序，对用户进行排序
Arrays.sort(per);
for(int i=0;i<per.length;i++){
for(int j=i+1;j<per.length;j++){
context.write(new Text(per[i]+"-"+per[j]+":"),new Text(friend));
}
}
}
}
/*
* @param Text 第一个是 reduce输入的key类型
* @param Text 第二个是 reduce输入的value类型
* @param Text 第三个是 reduce输出的key类型
* @param Text 第四个是 reduce输出的value类型
* */
public static class commonFriendReduce02 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 v:values){
sb.append(v).append(",");
}
sb.deleteCharAt(sb.length()-1);
context.write(key,new Text(sb.toString()));
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
// 设置job相关参数
Job job01 = Job.getInstance(conf);
// 设置jar、map、reduce相关类
job01.setJarByClass(CommonFriends.class);
job01.setMapperClass(commonFriendMap01.class);
job01.setReducerClass(commonFriendReduce01.class);
// 设置业务逻辑Reducer类输出key、value的数据类型
job01.setOutputKeyClass(Text.class);
job01.setOutputValueClass(Text.class);
// 指定输入输出路径
FileInputFormat.setInputPaths(job01,new Path("/mapreduce/src/main/resources/source"));
FileOutputFormat.setOutputPath(job01,new Path("/mapreduce/src/main/resources/output"));
Job job02 = Job.getInstance(conf);
job02.setJarByClass(CommonFriends.class);
job02.setMapperClass(commonFriendMap02.class);
job02.setReducerClass(commonFriendReduce02.class);
job02.setOutputKeyClass(Text.class);
job02.setOutputValueClass(Text.class);
//设定ReduceTask数量 一般等于Partitioner数量
job02.setNumReduceTasks(1);
// 指定输入输出路径
FileInputFormat.setInputPaths(job02,new Path("/mapreduce/src/main/resources/output"));
FileOutputFormat.setOutputPath(job02,new Path("/mapreduce/src/main/resources/final"));
//job提交
boolean res01 = job01.waitForCompletion(true);
boolean res02 = job02.waitForCompletion(true);
System.exit(res02&&res01?0:1);
}
}
最后的输出结果：
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
