MapReduce编程案例八

2019-08-27

Mapreduce经典案例两种方式求共同好友

题目描述

给出A-O个人中每个人的好友列表，求出哪些人两两之间有共同好友，以及他们的共同好友都有谁

数据样例：
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,E
A-C: D,F
A-D: E,F
A-E: B,C,D
A-F: B,C,D,E,O
A-G: C,D,E,F
A-H: C,D,E,O
A-I: O

方法一

处理逻辑：

实现思路：

实现共同好友分两步实现：
第一步：只求哪些人两两间有某个共同好友
1.将每组的user读到的第一个数据也就是冒号前面的值设置为value
2.将后面他的好友设置为key
- 一组数据A:B,C,D,F,E,O
- map输出reduce输入为：B A C A D A F A …
3.reduce循环遍历过来的value(friends)
- 将每组的数据加入到list中进行排序（默认排序）
- 然后将一组数据通过两层循环A拼接E，F,J然后从E开始继续，形成好友对
4.reduce端输出的key为好友对，value则为map端的key也就是friend
第二步：
- 将上次的结果作为输入，两两好友对作为key，拼接他们相同的value

处理逻辑代码

共同好友实现第一步

package cn.aparke.mr.friend;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
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.NullWritable;
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 CommonFriendsStepOne {
	
	public static class CommonFriendsOneMapper extends Mapper<LongWritable, Text, Text, Text>{
		Text k = new Text();
		Text v = new Text();
		
		//A:B,C,D,F,E,O
		// 输出：　B->A  C->A  D->A ...
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			String[] userAndFriends = value.toString().split(":");
			
			String user = userAndFriends[0];
			String[] friends = userAndFriends[1].split(",");
			v.set(user);
			for (String f : friends) {
				k.set(f);
				context.write(k, v);
			}
		}
	}
	
		
	public static class CommonFriendsOneReducer extends Reducer<Text, Text, Text, Text>{

		//传来一组数据reduce 处理
		// 一组数据：  B -->  A  E  F  J .....
		// 一组数据：  C -->  B  F  E  J .....
		@Override
		protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			
			//传来的key为friends value为用户
			ArrayList<String> userList = new ArrayList<>();
			
			//将用户添加到链表中进行排序，默认是字典顺序
			for(Text user: values){
				userList.add(user.toString());
			}
			/***
			 * java.util.Collections 是一个包装类。它包含有各种有关集合操作的静态多态方法。此类不能实例化，就像一个工具类，服务于Java的Collection框架。
 				1) 排序(Sort)使用sort方法可以根据元素的自然顺序 对指定列表按升序进行排序。列表中的所有元素都必须实现 Comparable 接口。
 				此列表内的所有元素都必须是使用指定比较器可相互比较的
			 */
			Collections.sort(userList);
			//一组数据通过两层循环A拼接E，F,J然后从E开始继续
			for(int i=0;i<userList.size()-1;i++){
				//外层循环为A
				for(int j=i+1;j<userList.size();j++){
					//里层循环为A后面的元素
					context.write(new Text(userList.get(i)+"-"+userList.get(j)), key);
				}
			}
			
		}
		
	}
	
	
public static void main(String[] args) throws Exception {

		
		Configuration conf = new Configuration();  
		
		Job job = Job.getInstance(conf);

		job.setJarByClass(CommonFriendsStepOne.class);

		job.setMapperClass(CommonFriendsOneMapper.class);
		job.setReducerClass(CommonFriendsOneReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		Path inpath=new Path("d:/testdata/friends/input");
		Path outpath=new Path("d:/testdata/friends/out_step1");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job,outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);
	}
}

共同好友实现第二步

package cn.aparke.mr.friend;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;

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.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;

/*
 * 上一步mapreduce的结果
B-C	A
B-D	A
B-F	A
B-G	A
B-H	A
B-I	A
B-K	A
B-O	A
C-D	A

让前面的好友对作为key reduce实现相同的key 将value拼接
 */
public class CommonFriendsStepTwo {

	public static class CommonFriendsTwoMapper extends Mapper<LongWritable, Text, Text, Text>{
		Text outKey = new Text();
		Text outVal = new Text();
		
		//输入B-D: A B-D E
		// 输出：　B-D:	A,E
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			String[] Friends = value.toString().split("\t");
			//好友对作为key
			String k = Friends[0];
			String v = Friends[1];
		
			outKey.set(k);
			outVal.set(v);
			context.write(outKey, outVal);
		}
	}
	
	public static class CommonFriendsTwoReducer extends Reducer<Text, Text, Text, Text>{

		Text outValue=new Text();
		//传来一组数据reduce 处理
		// 一组数据：  B-C	A
		@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 value : values) {
				sb.append(value.toString()).append(",");//循环追加每个value在key后面
			}
			sb.deleteCharAt(sb.length()-1);
	
			outValue.set(sb.toString());
			context.write(key, outValue);
			
		}
		
	}
public static void main(String[] args) throws Exception {

		
		Configuration conf = new Configuration();  
		
		Job job = Job.getInstance(conf);

		job.setJarByClass(CommonFriendsStepTwo.class);

		job.setMapperClass(CommonFriendsTwoMapper.class);
		job.setReducerClass(CommonFriendsTwoReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		Path inpath=new Path("d:/testdata/friends/out_step1");
		Path outpath=new Path("d:/testdata/friends/out_result");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job,outpath);
		

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

方法二

逻辑思路
此题旨在求两人之间的共同好友，原信息是<人，该人的所有好友>，因此首先以好友为键，人为值，交给reduce找出拥有此好友的所有人；再将这些人中两两配对作为键，之前的键（好友）作为值交给reduce去合并
简而言之我打算分成两个步骤，两次迭代
* 1）求出每一个人都是哪些人的共同好友  
* 2）把这些人（用共同好友的人）作为key,其好友作为value输出    
原文链接：https://blog.csdn.net/u012808902/article/details/77513188

一个代码实现两个job 需注意第一个job结束不能system.exit()

package cn.aparke.mr.friend;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
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;

	/**
	求所有两两用户之间的共同好友
	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

	输出结果为

	A-B:	C,E
	A-C:	D,F
	A-D:	E,F
	A-E:	B,C,D
	A-F:	B,C,D,E,O
	A-G:	C,D,E,F
	A-H:	C,D,E,O
	A-I:	O
	A-J:	B,O
	* @author aparke
	*
	*/
public class CommonFriends {
	/*
	 第一阶段的map函数主要完成以下任务
	 1.遍历原始文件中每行<所有朋友>信息
	 2.遍历“朋友”集合，以每个“朋友”为键，原来的“人”为值  即输出<朋友,人>
	 */
	static class CommonFriendsMapper01 extends Mapper<LongWritable, Text, Text, Text>{
		Text mapKey=new Text();
		Text mapVal=new Text();
		
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			//一组数据A:B,C,D,F,E,O
			String line = value.toString();
			String[] person_friends = line.split(":");
			String person = person_friends[0];//A
			mapVal.set(person);
			String[] friends = person_friends[1].split(",");//A的所有好友
			//遍历“朋友”集合，以每个“朋友”为键，原来的“人”为值  即输出<朋友,人>
			for(String friend : friends){
				mapKey.set(friend);
				//key为好友   value人
				context.write(mapKey,mapVal);
			}
		}
	}
	
	/*
	  第一阶段的reduce函数主要完成以下任务
	  1.对所有传过来的<朋友，list(人)>进行拼接，输出<朋友,拥有这名朋友的所有人>
	 */
	static class CommonFriendsReducer01 extends Reducer<Text, Text, Text, Text>{
		Text outVal=new Text();
		
		@Override
		protected void reduce(Text key, Iterable<Text> values,Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			
			for(Text friend : values){
				//循环遍历每个values ()
				sb.append(friend.toString()).append(",");
			}
			sb.deleteCharAt(sb.length()-1);//删除最后一个逗号
			
			outVal.set(sb.toString());
			//key不变，仍为某个好友，value 为拥有这名朋友的所有人
			context.write(key, outVal);
		}
	}

	/*
	第一阶段的结果
	A	I,K,C,B,G,F,H,O,D
	B	A,F,J,E
	C	A,E,B,H,F,G,K
	D	G,C,K,A,L,F,E,H
	*/
	
	/*
	第二阶段的map函数主要完成以下任务
	1.将上一阶段reduce输出的<朋友,拥有这名朋友的所有人>信息中的 “拥有这名朋友的所有人”进行排序 ，以防出现B-C C-B这样的重复
	2.将 “拥有这名朋友的所有人”进行两两配对，并将配对后的字符串当做键，“朋友”当做值输出，即输出<人-人，共同朋友>
	 */
	static class CommonFriendsMapper02 extends Mapper<LongWritable, Text, Text, Text>{
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			//一组数据来袭
			//A	I,K,C,B,G,F,H,O,D
			String line = value.toString();
			String[] friend_persons = line.split("\t");
			String friend = friend_persons[0];//A
			String[] persons = friend_persons[1].split(",");
			Arrays.sort(persons); //排序，默认按字典顺序，以防出现B-C C-B这样的重复
			
			//排序后两两配对     B,C,D,F,G,H,I,K,O
			for(int i=0;i<persons.length-1;i++){
				for(int j=i+1;j<persons.length;j++){
					context.write(new Text(persons[i]+"-"+persons[j]+":"), new Text(friend));
				}
			}
		}
	}
	
	/*
	 * 聚合，将相同key(人-人)的相同value 追加在一起，按照要求输出
	第二阶段的reduce函数主要完成以下任务
	1.<人-人，list(共同朋友)> 中的“共同好友”进行拼接 最后输出<人-人，两人的所有共同好友>
	 */
	static class CommonFriendsReducer02 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<String>();
			//将好友添加到集合中，去除相同的共同好友并排序
			for(Text friend : values){
				if(!set.contains(friend.toString()))
					set.add(friend.toString());
			}
			//对集合中的每个元素后面追加逗号
			for(String friend : set){
				sb.append(friend.toString()).append(",");
			}
			sb.deleteCharAt(sb.length()-1);//删除最后一个逗号
			
			context.write(key, new Text(sb.toString()));//key为人-人
		}
	}
	
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
 
		//第一阶段
		Job job1 = Job.getInstance(conf);
		job1.setJarByClass(CommonFriends.class);
		job1.setMapperClass(CommonFriendsMapper01.class);
		job1.setReducerClass(CommonFriendsReducer01.class);
		
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);
		
		FileInputFormat.setInputPaths(job1, new Path("D:\\testdata\\friends\\input"));
		FileOutputFormat.setOutputPath(job1, new Path("D:\\testdata\\friends\\output_temp"));
		
		boolean res1 = job1.waitForCompletion(true);
		
		//第二阶段
		Job job2 = Job.getInstance(conf);
		job2.setJarByClass(CommonFriends.class);
		job2.setMapperClass(CommonFriendsMapper02.class);
		job2.setReducerClass(CommonFriendsReducer02.class);
		
		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);
		
		FileInputFormat.setInputPaths(job2, new Path("D:\\testdata\\friends\\output_temp"));
		FileOutputFormat.setOutputPath(job2, new Path("D:\\testdata\\friends\\output"));
		
		boolean res2 = job2.waitForCompletion(true);
		
		System.exit(res1?0:1);
	}
}

项目地址

Mapreduce经典案例