大数据系列hadoop——MapReduce实例——好友推荐

逃离我推掉我的手 2022-05-17 08:54 191阅读 0赞

**目录**

一、简介

二、思路

三、实现

--------------------

# 一、简介 #

好友推荐功能简单的说是这样一个需求，预测某两个人是否认识，并推荐为好友。

# 二、思路 #

某两个非好友的用户，他们的共同好友越多，那么他们越可能认识。

比如，原始数据如下

Tom Cat Hello Hadoop Spring
    Cat Hello Spring
    Hello Tom Netty Hadoop Cat 
    Hadoop Tom Hello Netty Spring 
    Spring Tom Cat Hadoop
    Netty Hello Hadoop

每一行表示某个用户的好友列表，每一行的第一个名称为该用户的用户名，后面跟的用户名表示他的好友

那么我们需要得到如下的数据

Tom Netty 2
    Cat Netty 1
    Cat Hadoop 3
    Hello Spring 3
    Spring Netty 1

这里每一行的两个用户名即为预测的可能认识的两个用户，数字表示他们共有的好友数

map：

将每行的人名进行组合，最后输出key为人名组合，value为标志位。需要注意的是，人名需要按照一定顺序排序，防止同样的人名组合产生两个不同的key

标志位0，表示一定是直接好友，也就是用户本人和他用户列表中的每个用户是直接好友。

标志位1，表示可能是间接好友，因为每次只能处理一行，所以在某一行中两个人不是直接好友，但是在其他人的好友列表中，他们可能是直接好友，比如Cat和Hello，在第一行我们只能确定他们可能是间接好友，在第二行我们才知道他们其实是直接好友。然而，Cat和Hadoop才是真正的间接好友。因此这里需要设置为标志位，在reduce阶段才能确认是不是真正的间接好友。选择1为标志位也方便reduce阶段的累加计算。

reduce：

对每个人名组合进行处理。如果值存在标志位0，那么直接抛弃。如果标志位没有0，那么就能够确定肯定是间接好友，进行累加计算。

# 三、实现 #

@Slf4j
    public class RecommentFriendJob {
    	
    	/**
    	 * 
    	 * @param args 0|profile;1|input;2|output;3|master-ip;4|operator;5|homeDir
    	 * @throws Exception
    	 */
    	public static void main(String[] args) throws Exception {
    		
    		Configuration config = JobUtil.init(args);
    		
    		Job job = Job.getInstance(config);
    		
    		job.setJarByClass(RecommentFriendJob.class);
    		job.setJobName("recommentFriend");
    		
    		Path inputPath = new Path(args[1]);
    		FileInputFormat.addInputPath(job, inputPath);
    		
    		Path outputPath = new Path(args[2]);
    		if(outputPath.getFileSystem(config).exists(outputPath)) {
    			outputPath.getFileSystem(config).delete(outputPath, true);
    		}
    		FileOutputFormat.setOutputPath(job, outputPath);
    		
    		job.setMapperClass(ReconmentFriendMapper.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		job.setNumReduceTasks(1);
    		job.setReducerClass(RecommentFriendReducer.class);
    		
    		boolean isSuccess = job.waitForCompletion(true);
    		log.info("isSuccess:" + isSuccess);
    		System.exit(isSuccess ? 0 : 1);
    		
    	}
    	
    }

@Slf4j
    public class ReconmentFriendMapper extends Mapper<Object, Text, Text, IntWritable> {
    
    	/**
    	 * 
    	 * map的key为好友名，value为0|直接好友;1|可能是间接好友，需要在reduce中进行进一步处理
    	 * 
    	 */
    	@Override
    	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    		
    		log.info("key:" + key + ",value:" + value);
    		String[] friends = value.toString().split(" ");
    		for(int i=0; i < friends.length; i++) {
    			
    			String self = friends[i];
    			
    			for(int j=i+1; j < friends.length; j++) {
    				
    				log.info("i:" + i + ",j:" + j);
    				if(i == 0) {
    					// 直接好友
    					String directFriend = friends[j];
    					Text directFriendKey = new Text(sort(self, directFriend));
    					log.info("direct:" + directFriendKey.toString());
    					context.write(directFriendKey, new IntWritable(0));
    				} else {
    					// 可能是间接好友
    					String indirectFriend = friends[j];
    					Text indirectFriendKey = new Text(sort(self, indirectFriend));
    					log.info("indirect:" + indirectFriendKey.toString());
    					context.write(indirectFriendKey, new IntWritable(1));
    				}
    				
    			}
    			
    		}
    		
    	}
    	
    	private String sort(String self, String directFriend) {
    		if(self.compareToIgnoreCase(directFriend) < 0) {
    			return directFriend + " " + self;
    		}
    		return self + " " + directFriend;
    	}
    	
    }

@Slf4j
    public class RecommentFriendReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
    	@Override
    	public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    		log.info("key:" + key);
    		int sum = 0;
    		boolean isDirectFriend = false;
    		for(IntWritable value : values) {
    			if(value.get() == 0) {
    				// 直接好友
    				log.info("direct friend");
    				isDirectFriend = true;
    				break;
    			}
    			sum = sum + value.get();
    		}
    		
    		if(!isDirectFriend) {
    			context.write(key, new IntWritable(sum));
    		}
    	}
    	
    }

[https://github.com/wulinfeng2/hadoopDemo][https_github.com_wulinfeng2_hadoopDemo]

[https_github.com_wulinfeng2_hadoopDemo]: https://github.com/wulinfeng2/hadoopDemo