MapReduce数据分析(6)共同好友

mac2026-08-09  3

六、MapReduce第六讲共同好友(Common friends)

某某社交网站,有如下用户好友关系:

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

数据解释:

A:B,C,D,F,E,O

每行数据以冒号为分隔符:

1、冒号左边是网站的一个用户A; 2、冒号右边是用户A的好友列表(各好友间以逗号为分隔符); 现在,需要对网站的几十亿用户进行分析,找出哪些用户两两之间有共同好友,以及他俩的共同好友都有哪些人。比如,A、B两个用户拥有共同好友C和E;

最终统计的结果数据示意如下:

A-L F,E,D A-M E,F B-C A B-D A,E B-E C

设计思路: 整个实验过程需要写两个MapReduce程序来实现。 第一个job实现出来的结果是

B-C A B-D A

该数据的意思是B和C用户、B和D用户都是A用户的好友 第二个job实现出来的结果是

A-B E,C A-C D,F A-D E,F

该数据的意思是A和B的共同好友是E用户,C用户。

接下来我们开始写第一个job类。 代码如下:

package demo; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; public class demo{ public static void main(String[] args) { Configuration conf=new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(demo.class); job.setMapperClass(MMapper.class); job.setReducerClass(MReduce.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])); job.waitForCompletion(true); } public static class MMapper extends Mapper<LongWritable, Text, Text, Text>{ // 输入数据形式如: A:B,C,D,F,E,O protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException { // 转换数据类型并用:切割 String[] line = value.toString().split(":"); //得到数据中的用户 String user = line[0]; String[] itmse = line[1].split(","); //得到数据中的好友 for (String v : itmse) { //将每一个好友作为key,用户作为value context.write(new Text(v), new Text(user)); } } } public static class MReduce extends Reducer<Text, Text, Text, Text>{ protected void reduce(Text key, Iterable<Text> value, Context context) throws IOException, InterruptedException { ArrayList<Text> arrayList = new ArrayList<Text>(); // 将这一组拥有共同好友f的user们从迭代器中取出,放入一个arraylist暂存 for (Text u : value) { arrayList.add(new Text(u)); } //对users排个序,以免拼俩俩对时出现A-F 又有F-A的现象 Collections.sort(arrayList); // 把这一对user进行两两组合,并将: //1.组合作为key //2.共同的好友f作为value //返回给reduce task作为本job的最终结果 for (int i = 0; i < arrayList.size()-1; i++) { for (int j = i+1; j <arrayList.size(); j++) { context.write(new Text(arrayList.get(i)+"-"+arrayList.get(j)), key); } } } } }

运行结果:

下面我们开始写第二个job

代码如下;

package xxx; 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 demo{ public static void main(String[] args) throws Exception { Configuration conf=new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(demo.class); job.setMapperClass(MMapper.class); job.setReducerClass(MReduce.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])); job.waitForCompletion(true); } public static class MMapper extends Mapper<LongWritable, Text, Text, Text>{ // 输入数据形式如: A:B,C,D,F,E,O protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException { // 将数据按制表符切分 String[] line = value.toString().split("\t"); // 将切出来的B-C用户对作为key,共同好友A作为value // 返回给map task context.write(new Text(line[0]), new Text(line[1])); } } public static class MReduce extends Reducer<Text, Text, Text, Text>{ protected void reduce(Text key, Iterable<Text> value, Context context) throws IOException, InterruptedException { // 构造一个StringBuilder用于拼接字符串 StringBuffer sb = new StringBuffer(); for (Text v : value) { // 将这个用户对的所有共同好友拼接在一起 sb.append(v).append(","); } //用户对作为key,拼接好的所有共同好友作为value context.write(key, new Text(sb.substring(0,sb.length()-1))); } } }

运行结果:

本次教程就到次结束了,有什么不懂的欢迎大家下方评论。博主都会帮你解答的。觉的写的不错的点个赞价格关注,多多支持。

最新回复(0)