MapReduce 社交好友推薦算法

原理

如果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文件。

1. cd /data/mydata  
2. vim friend_data.txt  

2.將如下初始數據寫入其中（注意數據之間以空格分割）

1. A B  
2. C D  
3. E F  
4. F G  
5. B D  
6. B C  

3.切換到/apps/hadoop/sbin目錄下，開啟Hadoop相關進程

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

4.輸入JPS查看一下相關進程是否已經啟動。

1. jps  

5.在HDFS的根下創建一個friend目錄，並將friend_data.txt文件上傳到HDFS上的friend文件夾下。

1. hadoop fs -mkdir /friend  
2. 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類的代碼。

完整代碼為：

1. package mr_friend;  
2. import java.io.IOException;  
3. import java.net.URI;  
4. import java.net.URISyntaxException;  
5. import java.util.HashSet;  
6. import java.util.Iterator;  
7. import java.util.Set;  
8.     
9. import org.apache.hadoop.conf.Configuration;  
10. import org.apache.hadoop.fs.FileSystem;  
11. import org.apache.hadoop.fs.Path;  
12. import org.apache.hadoop.io.LongWritable;  
13. import org.apache.hadoop.io.Text;  
14. import org.apache.hadoop.mapreduce.Job;  
15. import org.apache.hadoop.mapreduce.Mapper;  
16. import org.apache.hadoop.mapreduce.Reducer;  
17. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
19.     
20. public class Find_Friend {  
21. /* 
22. map結果： 
23. A B 
24. B A 
25. C D 
26. D C 
27. E F 
28. F E 
29. F G 
30. G F 
31. B D 
32. D B 
33. B C 
34. C B 
35. */  
36.     
37.     public static class FindFriendsMapper extends Mapper<LongWritable, Text, Text, Text> {  
38.         @Override  
39.         protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)  
40.                 throws IOException, InterruptedException {  
41.             String line = value.toString();  
42.             String array[] = line.split("\\s+");  
43.             context.write(new Text(array[0]), new Text(array[1]));  
44.             context.write(new Text(array[1]), new Text(array[0]));  
45.         }  
46.     }  
47.     
48. /* 
49. map之后，Shuffling將相同key的整理在一起，結果如下： 
50. shuffling結果(將結果輸出到reduce)： 
51. A B 
52.    
53. B A 
54. B D 
55. B C 
56.    
57. C D 
58. C B 
59.    
60. E F 
61.    
62. F E 
63. F G 
64.    
65. G F 
66. */  
67. //reduce將上面的數據進行笛卡爾積計算  
68.     public static class FindFriendsReduce extends Reducer<Text, Text, Text, Text> {  
69.         @Override  
70.         protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)  
71.     throws IOException, InterruptedException {  
72.     //將重復數據去重  
73.     Set<String> set = new HashSet<String>();  
74.         for (Text v : values) {  
75.         set.add(v.toString());  
76.         }  
77.     
78.         if (set.size() > 1) {  
79.         for (Iterator<String> i = set.iterator(); i.hasNext();) {  
80.             String qqName = i.next();  
81.             for (Iterator<String> j = set.iterator(); j.hasNext();) {  
82.                 String otherQqName = j.next();  
83.                 if (!qqName.equals(otherQqName)) {  
84.                 context.write(new Text(qqName), new Text(otherQqName));  
85.                 }  
86.                 }  
87.                 }  
88.                 }  
89.                 }  
90.                 }  
91.     
92.                 public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {  
93.                 final String INPUT_PATH = "hdfs://127.0.0.1:9000/friend/friend_data.txt";  
94.                 final String OUTPUT_PATH = "hdfs://127.0.0.1:9000/friend/output";  
95.     
96.                 Configuration conf = new Configuration();  
97.                 //Configuration：map/reduce的配置類，向hadoop框架描述map-reduce執行的工作  
98.     
99.                 final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);  
100.                 if(fileSystem.exists(new Path(OUTPUT_PATH))) {  
101.                 fileSystem.delete(new Path(OUTPUT_PATH), true);  
102.                 }  
103.     
104.                 Job job = Job.getInstance(conf, "Find_Friend");//設置一個用戶定義的job名稱  
105.                 job.setJarByClass(Find_Friend.class);  
106.                 job.setMapperClass(FindFriendsMapper.class);    //為job設置Mapper類  
107.                 job.setReducerClass(FindFriendsReduce.class);    //為job設置Reducer類  
108.                 job.setOutputKeyClass(Text.class);              //為job的輸出數據設置Key類  
109.                 job.setOutputValueClass(Text.class);            //為job輸出設置value類  
110.     
111.                 FileInputFormat.addInputPath(job, new Path(INPUT_PATH));  
112.                 FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));  
113.     
114.                 System.exit(job.waitForCompletion(true) ?0 : 1);          //運行job  
115.                 }  
116.     
117.                 }  

11.下面在Find_Friend類下，單擊右鍵，選擇Run As=>Run on Hadoop，運行程序，查看執行結果。

12.程序執行完以后，查看HDFS上的/friend/output目錄中的計算結果。

1. hadoop fs -ls -R /friend  
2. hadoop fs -cat /friend/output/part-r-00000  

通過分析結果，就得出了各位同學的可能認識的人的列表了。

至此，實驗就已經結束了。