MapReduce 社交好友推荐算法 - HelloWorld开发者社区

原理

如果A和B具有好友关系，B和C具有好友关系，而A和C却不是好友关系，那么我们称A和C这样的关系为：二度好友关系。

在生活中，二度好友推荐的运用非常广泛，比如某些主流社交产品中都会有"可能认识的人"这样的功能，一般来说可能认识的人就是通过二度好友关系搜索得到的，在传统的关系型数据库中，可以通过图的广度优先遍历算法实现，而且深度限定为2，然而在海量的数据中，这样的遍历成本太大，所以有必要利用MapReduce编程模型来并行化。

初始数据如下：

A B

C D

E F

F G

B D

B C

map阶段得到的结果为：

Key：A Value：B

Key：B Value：A C D

Key：C Value：B D

Key：E Value：F

Key：F Value：E G

Key：G Value：F

Reduce阶段再将Value进行笛卡尔积运算就可以得到二度好友关系了

（笛卡尔积公式：A×B={(x,y)|x∈A∧y∈B}

例如，A={a,b}, B={0,1,2}，则

A×B={(a, 0), (a, 1), (a, 2), (b, 0), (b, 1), (b, 2)}

B×A={(0, a), (0, b), (1, a), (1, b), (2, a), (2, b)}）

环境

Linux Ubuntu 14.04

jdk-7u75-linux-x64

Hadoop 2.6.0-cdh5.4.5

内容

通过初始数据，假设有A、B、C、D、E、F、G七位同学，其中A与B是好友关系，C与D是好友关系，E与F是好友关系，F与G是好友关系，B与D是好友关系，B与C是好友关系，通过分析A与B是好友，且B与C也是好友，我们就认为A与C互为可能认识的人，向A与C互相推荐对方。

实验步骤

1.首先，来准备实验需要用到的数据，切换到/data/mydata目录下，使用vim编辑一个friend_data.txt文件。

cd /data/mydata
vim friend_data.txt

2.将如下初始数据写入其中（注意数据之间以空格分割）

3.切换到/apps/hadoop/sbin目录下，开启Hadoop相关进程

cd /apps/hadoop/sbin
./start-all.sh

4.输入JPS查看一下相关进程是否已经启动。

5.在HDFS的根下创建一个friend目录，并将friend_data.txt文件上传到HDFS上的friend文件夹下。

hadoop fs -mkdir /friend
hadoop fs -put /data/mydata/friend_data.txt /friend

6.打开Eclipse，创建一个Map/Reduce项目。

7.设置项目名为mr_sf并点击Finish。

8.创建一个包，名为mr_friend。

9.创建一个类，名为Find_Friend。

10.下面开始编写Find_Friend类的代码。

完整代码为：

package mr_friend;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
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;
public class Find_Friend {
/*
map结果：
A B
B A
C D
D C
E F
F E
F G
G F
B D
D B
B C
C B
*/
public static class FindFriendsMapper extends Mapper<LongWritable, Text, Text, Text> {
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
String line = value.toString();
String array[] = line.split("\\s+");
context.write(new Text(array[0]), new Text(array[1]));
context.write(new Text(array[1]), new Text(array[0]));
}
}
/*
map之后，Shuffling将相同key的整理在一起，结果如下：
shuffling结果(将结果输出到reduce)：
A B
B A
B D
B C
C D
C B
E F
F E
F G
G F
*/
//reduce将上面的数据进行笛卡尔积计算
public static class FindFriendsReduce extends Reducer<Text, Text, Text, Text> {
@Override
protected void reduce(Text key, Iterable values, Reducer<Text, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
//将重复数据去重
Set set = new HashSet();
for (Text v : values) {
set.add(v.toString());
}
if (set.size() > 1) {
for (Iterator i = set.iterator(); i.hasNext();) {
String qqName = i.next();
for (Iterator j = set.iterator(); j.hasNext();) {
String otherQqName = j.next();
if (!qqName.equals(otherQqName)) {
context.write(new Text(qqName), new Text(otherQqName));
}
}
}
}
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
final String INPUT_PATH = "hdfs://127.0.0.1:9000/friend/friend_data.txt";
final String OUTPUT_PATH = "hdfs://127.0.0.1:9000/friend/output";
Configuration conf = new Configuration();
//Configuration：map/reduce的配置类，向hadoop框架描述map-reduce执行的工作
final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
if(fileSystem.exists(new Path(OUTPUT_PATH))) {
fileSystem.delete(new Path(OUTPUT_PATH), true);
}
Job job = Job.getInstance(conf, "Find_Friend");//设置一个用户定义的job名称
job.setJarByClass(Find_Friend.class);
job.setMapperClass(FindFriendsMapper.class); //为job设置Mapper类
job.setReducerClass(FindFriendsReduce.class); //为job设置Reducer类
job.setOutputKeyClass(Text.class); //为job的输出数据设置Key类
job.setOutputValueClass(Text.class); //为job输出设置value类
FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
System.exit(job.waitForCompletion(true) ?0 : 1); //运行job
}
}