MapReduce实现二度好友关系

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了MapReduce实现二度好友关系相关的知识,希望对你有一定的参考价值。

一、问题定义

     我在网上找了些,关于二度人脉算法的实现,大部分无非是通过广度搜索算法来查找,犹豫深度已经明确了2以内;这个算法其实很简单,第一步找到你关注的人;第二步找到这些人关注的人,最后找出第二步结果中出现频率最高的一个或多个人(频率这块没完成),即完成。

     但如果有千万级别的用户,那在运算时,就肯定会把这些用户的follow 关系放到内存中,计算的时候依次查找;先说明下我没有明确的诊断对比,这样做的效果一定没 基于hadoop实现的好;只是自己,想用hadoop实现下,最近也在学;若有不足的地方还请指点。

  任务是求其其中的二度人脉、潜在好友,也就是如下图:

技术分享

  比如I认识C、G、H,但C不认识G,那么C-G就是一对潜在好友,但G-H早就认识了,因此不算为潜在好友。

  那么一个关键问题是如何输入输入。

  首先是五项五环图,可以看出共有13条边,那么输入数据也有13条就够了,比如说先输入AB,那么轮到b时候就不输入BA了,级变速如也没关系,因为会去重。

 

二、原理分析

  首先,我们进行第一个MapReduce,同样是一个输入行,产生一对互逆的关系,压入context,例如Tom Lucy这个输入行就在Map阶段搞出Tom Lucy-Lucy Tom这样的互逆关系。之后Map-reduce会自动对context中相同的key合并在一起。例如由于存在Tom Lucy、Tom Jack,显然会产生一个Tom:{Lucy,Jack},这是Reduce阶段以开始的键值对。这个键值对相当于Tom所认识的人。先进行如下的输出,潜在好友显然会在{Lucy,Jack}这个Tom所认识的人产生,对这个数组做笛卡尔乘积,形成关系:{<Lucy,Lucy>,<Jack,Jack>,<Lucy,Jack>,<Jack,Lucy>},也就是<Lucy,Lucy>这类无意义的剔除,<Lucy,Jack>,<Jack,Lucy>认定为一个关系,将剩余关系进行如下的输出。

  不过计算笛卡尔积就像双重for对同一个数组,重复计算了一半,怎么减少了,我程序里是HashSet,第二重如何从第一宠Set的iterator哪里开始呢。

三、代码

3.1 Mapper

package friends;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class Deg2FriendMapper extends Mapper<LongWritable, Text, Text, Text> {

	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		String line = value.toString();
		//   "\t"表示制表符
		//StringTokenizer st = new StringTokenizer(line,",");
		//while(st.hasMoreTokens())
		//用while循环的时候是一行有很多才需要
		String[] ss = line.split(",");
		context.write(new Text(ss[0]), new Text(ss[1]));
		context.write(new Text(ss[1]), new Text(ss[0]));
	}

}

  

3.2 Reducer

package friends;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class Deg2Reducer extends Reducer<Text, Text, Text, Text> {

	public void reduce(Text key, Iterable<Text> value, Context context)
			throws IOException, InterruptedException {
		// process values
		
		//首先是key相同的合并,同时取出value笛卡尔积之后的重复关系
		Set<String> set = new HashSet<String>();
		
		for (Text t : value) {//相同key合并
			//但是为什么用HashSet,因为Map里面谢了反响关系,比如 对于A节点,谢了AB,BA,
			//对于B节点,谢了BA,AB,那么A开头的有两次AB,去重,
			//为什么要for循环 因为A可能有很多朋友
			//
			set.add(t.toString());
		}
		if(set.size()>=2) {//否则说明只有一度好友关系
			//对value的值做笛卡尔积
			Iterator<String> iter = set.iterator();
			while(iter.hasNext()) {
				String name = iter.next();
				//iterator写成for循环的话 第三个条件没有 否则for内娶不到元素
				for(Iterator<String> iter2 = set.iterator();iter2.hasNext();) {
					String name2 = iter2.next();
					if(!name2.equals(name)) {//相同元素不算关系
						context.write(new Text(name), new Text(name2));
					}
				}
			}
			
		}
	}

}

  

3.2 Main

package friends;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class Deg2Main {

	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		Configuration conf = new Configuration(); //对应于mapred-site.xml
		Job job = new Job(conf,"Deg2MR");
		job.setJarByClass(Deg2Main.class);
		job.setMapperClass(Deg2FriendMapper.class);
		job.setReducerClass(Deg2Reducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		job.setNumReduceTasks(1);
		//"/in"解析不了  提示文件不存在 因为把他们认为是本地文件了 因为有个 file:/
		FileInputFormat.addInputPath(job, new Path("hdfs://192.168.58.180:8020/MLTest/Deg2MR/Deg2MR.txt"));
		//输出文件不能存在   
		FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.58.180:8020/MLTest/Deg2MR/Deg2Out"));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

  

3.4 日志

m:org.apache.hadoop.mapreduce.Job.updateStatus(Job.java:323)
  INFO - Job job_local1127799899_0001 completed successfully
 DEBUG - PrivilegedAction as:hxsyl (auth:SIMPLE) from:org.apache.hadoop.mapreduce.Job.getCounters(Job.java:765)
  INFO - Counters: 38
	File System Counters
		FILE: Number of bytes read=740
		FILE: Number of bytes written=509736
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=132
		HDFS: Number of bytes written=206
		HDFS: Number of read operations=13
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=4
	Map-Reduce Framework
		Map input records=13
		Map output records=26
		Map output bytes=106
		Map output materialized bytes=164
		Input split bytes=116
		Combine input records=0
		Combine output records=0
		Reduce input groups=10
		Reduce shuffle bytes=164
		Reduce input records=26
		Reduce output records=50
		Spilled Records=52
		Shuffled Maps =1
		Failed Shuffles=0
		Merged Map outputs=1
		GC time elapsed (ms)=3
		CPU time spent (ms)=0
		Physical memory (bytes) snapshot=0
		Virtual memory (bytes) snapshot=0
		Total committed heap usage (bytes)=456130560
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=66
	File Output Format Counters 
		Bytes Written=206
 DEBUG - PrivilegedAction as:hxsyl (auth:SIMPLE) from:org.apache.hadoop.mapreduce.Job.updateStatus(Job.java:323)
 DEBUG - stopping client from cache: [email protected]
 DEBUG - removing client from cache: [email protected]
 DEBUG - stopping actual client because no more references remain: [email protected]
 DEBUG - Stopping client
 DEBUG - IPC Client (521081105) connection to /192.168.58.180:8020 from hxsyl: closed
 DEBUG - IPC Client (521081105) connection to /192.168.58.180:8020 from hxsyl: stopped, remaining connections 0
 

  

3.5 输出

B	H
H	B
A	C
C	A
B	D
B	F
B	I 
D	B
D	F
D	I 
F	B
F	D
F	I 
I 	B
I 	D
I 	F
C	E
C	F
E	C
E	F
F	C
F	E
D	F
F	D
C	D
C	E
C	G
D	C
D	E
D	G
E	C
E	D
E	G
G	C
G	D
G	E
F	H
F	I
H	F
H	I
I	F
I	H
A	G
A	I
G	A
G	I
I	A
I	G
G	H
H	G

  

四、思考

4.1 单向

  类似父子关系找爷孙关系,或者是关注关系或者follow关系,那么Mapper阶段不相互存入就可。

4.2 你最受欢迎的二度人脉

  简单描述:即你关注的人中有N个人同时都关注了 XXX 。

4.3 Set遍历

  双重iterator便利HashSet,第二重如何从第一宠Set的iterator哪里开始呢。这样可以少算一倍,应该可以吧set转为数数组吧。

  不过这样也好,A是B的二度,那么B也是A的二度....

4.4 另外

  一开始reducer里写错了,set.add(toString.toString()),竟然没报错,没有toString这个变量。然后日志是reducer阶段没有任何写入。

五、参考文献

  http://blog.csdn.net/yongh701/article/details/50630498

  http://blog.csdn.net/u013926113/article/details/51539306

  https://my.oschina.net/BreathL/blog/75112

以上是关于MapReduce实现二度好友关系的主要内容,如果未能解决你的问题,请参考以下文章

经典教程 | 基于Spark GraphX实现微博二度关系推荐

MapReduce实现好友推荐

深度和广度优先搜索算法

深度和广度优先搜索算法

MapReduce课程设计——好友推荐功能

朋友圈的人脉关系的算法