《Hadoop进阶》利用Hadoop构建豆瓣图书推荐系统

转载请注明出处： 转载自  Thinkgamer的CSDN博客：blog.csdn.net/gamer_gyt代码下载地址：点击查看1：推荐系统概述2：需求分析：推荐系统的指标设计3：算法模型：基于物品的协同过滤并行算法设计4：架构设计：推荐系统架构5：程序实现：MR2V程序实现6：推荐系统评估

一、推荐系统概述

       推荐系统广泛存在与各大网站上，比如说亚马逊，淘宝等电商类网站上，而且在社交网络上也应用的十分广泛，比如说facebook的你可能认识，微博的好友推荐，也比如说csdn博客的你可能喜欢等等。这是我目前在做的一个豆瓣图书推荐系统，采用的算法主要是协同过滤算法，使用Python+Django+Mysql进行部署github地址：点击查看推荐算法的分类主要包括：按数据使用划分：协同过滤算法：UserCF, ItemCF, ModelCF基于内容的推荐: 用户内容属性和物品内容属性社会化过滤：基于用户的社会网络关系按模型划分：最近邻模型:基于距离的协同过滤算法Latent Factor Mode(SVD)：基于矩阵分解的模型Graph：图模型，社会网络图模型基于用户和基于Item的协同过滤算法，大家可以参考我之前写的一篇博客：http://blog.csdn.net/gamer_gyt/article/details/51346159里边详细介绍了基于用户的协同过滤算法和基于Item的协同过滤算法，以及他们的python实现版本，在这里就不过多进行论述

二、需求分析：豆瓣图书推荐系统指标设计

      推荐系统是为了更精准的为用户推荐他们想要的内容，如果一个用户在浏览图书信息的时候，通过对用户数据的记录，和已经存在的其他的用户记录进行分析，从而为用户推荐相应的数据数据集来源为，豆瓣图书，采用python爬虫脚本爬取相应的信息如下（以下信息为我预处理之后存入Mysql数据库的规整信息）user对book的打分表：这里我采用的是基于Item的协同过滤算法，通过评分来计算用户可能对书本的评分，过滤掉用户已经看过的，选择剩余的TopK推荐给用户。

三、算法模型：基于物品的协同过滤并行算法设计

现在我们以user对book的打分表为计算的数据集，首先导出生成score.txt文件每行三个字段，分别是userid，socre，bookid（每个字段之间以\t分割）并行算法的实现思想    (1)：建立物品的同现矩阵    (2)：建立用户对物品的评分矩阵    (3)：矩阵计算法推荐结果为了方便，我们拿一个小的数据集进行说明，数据样例如下：

1,101,5.0
1,102,3.0
1,103,2.5
2,101,2.0
2,102,2.5
2,103,5.0
2,104,2.0
3,101,2.0
3,104,4.0
3,105,4.5
3,107,5.0
4,101,5.0
4,103,3.0
4,104,4.5
4,106,4.0
5,101,4.0
5,102,3.0
5,103,2.0
5,104,4.0
5,105,3.5
5,106,4.0

（1）：建立物品的同现矩阵

按用户分组，找到每个用户所选的物品，单独出现计数及两两一组计数。

[101] [102] [103] [104] [105] [106] [107]
[101]   5     3     4     4     2     2     1
[102]   3     3     3     2     1     1     0
[103]   4     3     4     3     1     2     0
[104]   4     2     3     4     2     2     1
[105]   2     1     1     2     2     1     1
[106]   2     1     2     2     1     2     0
[107]   1     0     0     1     1     0     1

（2）：建立用户对物品的评分矩阵

       U3[101] 2.0[102] 0.0[103] 0.0[104] 4.0[105] 4.5[106] 0.0[107] 5.0

（3）：矩阵计算法推荐结果

同现矩阵*评分矩阵=推荐结果

四、架构设计：推荐系统架构

系统架构设计如下：App Service，Applicate业务系统，HDFS为分布式文件系统，Mapreduce为运行MRV2程序业务系统记录了用户的行为和对物品的打分设置系统定时器CRON，定时，增量向HDFS导入数据(userid,itemid,value,time)。完成导入后，设置系统定时器，启动MapReduce程序，运行推荐算法。完成计算后，设置系统定时器，从HDFS导出推荐结果数据到数据库，方便以后的及时查询。

五、程序实现：MR2V程序实现

新建Java类:   bookRecommend.java,主任务启动程序   Step1.java,按用户分组,计算所有物品出现的组合列表,得到用户对物品的评分矩阵   Step2.java,对物品组合列表进行计数,建立物品的同现矩阵   Step3.java,对同现矩阵和评分矩阵转型   Step4.java,合并矩阵,并计算推荐结果列表   HdfsGYT.java,HDFS操作工具类我们拿上边的数据集为例bookRecommend.java

package bookTuijian;import java.io.IOException;import java.net.URISyntaxException;import java.util.HashMap;import java.util.Map;import java.util.regex.Pattern;public class bookRecommend {/*** @param args* 驱动程序，控制所有的计算结果*/public static final String  HDFS = "hdfs://127.0.0.1:9000";public static final Pattern DELIMITER = Pattern.compile("[\t,]");public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {// TODO Auto-generated method stubMap<String,String>  path = new HashMap<String,String>();path.put("local_file", "MyItems/bookTuijian/uid_to_bid.csv");          //本地文件所在的目录path.put("hdfs_root_file", HDFS+"/mr/bookRecommend/uid_to_bid");       //上传本地文件到HDFS上的存放路径path.put("hdfs_step1_input", path.get("hdfs_root_file"));       //step1的输入文件存放目录path.put("hdfs_step1_output", HDFS+"/mr/bookRecommend/step1"); //hdfs上第一步运行的结果存放文件目录path.put("hdfs_step2_input", path.get("hdfs_step1_output"));           //step2的输入文件目录path.put("hdfs_step2_output", HDFS+"/mr/bookRecommend/step2");      //step2的输出文件目录path.put("hdfs_step3_1_input", path.get("hdfs_step1_output"));        //构建评分矩阵path.put("hdfs_step3_1_output", HDFS+"/mr/bookRecommend/Step3_1");path.put("hdfs_step3_2_input", path.get("hdfs_step2_output"));        //构建同现矩阵path.put("hdfs_step3_2_output", HDFS+"/mr/bookRecommend/Step3_2");path.put("hdfs_step4_input_1", path.get("hdfs_step3_1_output"));    //计算乘积path.put("hdfs_step4_input_2", path.get("hdfs_step3_2_output"));path.put("hdfs_step4_output", HDFS+"/mr/bookRecommend/result");Step1.run(path);     //Step2.run(path);Step3_1.run(path);   //构造评分矩阵Step3_2.run(path);   //构造同现矩阵Step4.run(path);       //计算乘积System.exit(0);}}

Step1.java运行结果：               Step2.java运行结果：         Step3_1.java运行结果：         Step3_2.java运行结果：             Step4.java最终结果为（第一列是userid，第二列是itemid，第三列是喜欢程序）：                  最终需要从中过滤掉用户已经看过的书籍，从而将余下的排序取Top K进行推荐eg：用户 3，他对101，102，103，104，105，106，107的兴趣程度为，42.5，20.0，26.5，40.0，27.0，17.5，16.0，排序后item的id顺序为 101，104，105，103，102，106，107，他已经打过分的有101，104，105，107，则余下的按顺序为103，102，106加入这里要给用户3推荐2个的话，那么就会推荐103和102最终运行结果hdfs截图如下：                                    接下来就是修改部分实验代码为我所用了，因为数据集的不同和数据类型的不一致，所以我们要灵活的运用代码，说一下我在这里遇到的问题吧：        因为运行上边的样例文件时，并没有报错，是因为，在矩阵乘法阶段，也就是Step4.java，Map类中需要先构造同现矩阵coocurenceMatrix，如果不构造同现矩阵，那么在进行乘法时将会报错，这就是我运行score.txt文件时，所遇到的问题，因为hadoop从hdfs上读取小文件时，会先读占用空间大的文件，这样就不难保证先生成coocurenceMatrix了，so在这里我们需要进行将代码进行改善(Step4.java进行修改，这里把矩阵乘法进行分开计算，先进行对于位置相乘Step4_Updata.java，最后进行加法Step4_Updata2.java)Step4_Updata.java:

package bookTuijian;import java.io.IOException;import java.net.URISyntaxException;import java.util.HashMap;import java.util.Iterator;import java.util.Map;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.InputSplit;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.input.FileSplit;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;/** 是对Step4的优化，分为矩阵相乘和相加，这一步是相乘*/public class Step4_Updata {public static class Step4_Updata_Map extends Mapper< LongWritable, Text, Text, Text>{String filename;@Overrideprotected void setup(Context context) throws IOException,InterruptedException {// TODO Auto-generated method stubInputSplit input = context.getInputSplit();filename = ((FileSplit) input).getPath().getParent().getName();System.out.println("FileName：" +filename);}@Overrideprotected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {// TODO Auto-generated method stubString[] tokens = bookRecommend.DELIMITER.split(value.toString());      //切分if(filename.equals("Step3_2") ){ //同现矩阵String[] v1 = tokens[0].split(":");String itemID1 = v1[0];String itemID2 = v1[1];String num = tokens[1];Text key1 = new Text(itemID1);Text value1 = new Text("A:" + itemID2 +"," +num);context.write(key1,value1);//				System.out.println(key1.toString() + "\t" + value1.toString());}else{    //评分矩阵String[] v2 = tokens[1].split(":");String itemID = tokens[0];String userID = v2[0];String score = v2[1];Text key1 = new Text(itemID);Text value1 = new Text("B:" + userID + "," + score);context.write(key1,value1);//				System.out.println(key1.toString() + "\t" + value1.toString());}}}public static class Step4_Updata_Reduce extends Reducer<Text, Text, Text, Text>{@Overrideprotected void reduce(Text key, Iterable<Text> values,Context context)	throws IOException, InterruptedException {// TODO Auto-generated method stub//			System.out.println(key.toString()+ ":");Map mapA = new HashMap();Map mapB = new HashMap();for(Text line : values){String val = line.toString();//				System.out.println(val);if(val.startsWith("A")){String[] kv = bookRecommend.DELIMITER.split(val.substring(2));mapA.put(kv[0], kv[1]); //ItemID, num//					System.out.println(kv[0] + "\t" + kv[1] + "--------------1");}else if(val.startsWith("B")){String[] kv = bookRecommend.DELIMITER.split(val.substring(2));mapB.put(kv[0], kv[1]); //userID, score//					System.out.println(kv[0] + "\t" + kv[1] + "--------------2");}}double result = 0;Iterator iterA = mapA.keySet().iterator();while(iterA.hasNext()){String mapkA = (String) iterA.next();  //itemIDint num = Integer.parseInt((String) mapA.get(mapkA));     // numIterator iterB = mapB.keySet().iterator();while(iterB.hasNext()){String mapkB = (String)iterB.next(); //UserIDdouble score = Double.parseDouble((String) mapB.get(mapkB));  //scoreresult = num * score; //矩阵乘法结果Text key2 = new Text(mapkB);Text value2 = new Text(mapkA + "," +result);context.write(key2,value2); //userID \t  itemID,result//					System.out.println(key2.toString() + "\t" + value2.toString());}}}}public static void run(Map<String, String> path) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {// TODO Auto-generated method stubString input_1 = path.get("hdfs_step4_updata_input");String input_2 = path.get("hdfs_step4_updata2_input");String output = path.get("hdfs_step4_updata_output");hdfsGYT hdfs = new hdfsGYT();hdfs.rmr(output);Job job = new Job(new Configuration(), "Step4_updata");job.setJarByClass(Step4_Updata.class);//设置文件输入输出路径FileInputFormat.setInputPaths(job, new Path(input_1),new Path(input_2));FileOutputFormat.setOutputPath(job, new Path(output));//设置map和reduce类job.setMapperClass(Step4_Updata_Map.class);job.setReducerClass(Step4_Updata_Reduce.class);//设置Map输出job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);//设置Reduce输出job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);//设置文件输入输出job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);job.waitForCompletion(true);}}

Step4_Updata2.java

package bookTuijian;import java.io.IOException;import java.net.URISyntaxException;import java.util.HashMap;import java.util.Iterator;import java.util.Map;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.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;public class Step4_Updata2 {public static class Step4_Updata2_Map extends Mapper<LongWritable, Text, Text, Text>{@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// TODO Auto-generated method stubString[] tokens = bookRecommend.DELIMITER.split(value.toString());Text key1 = new Text(tokens[0]);//userIDText value1 = new Text(tokens[1] + "," + tokens[2]);context.write(key1, value1);    //itemID,result}}public static class Step4_Updata_Reduce extends Reducer< Text, Text, Text, Text>{@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {// TODO Auto-generated method stubMap map = new HashMap();for(Text line: values){System.out.println(line.toString());String[] tokens = bookRecommend.DELIMITER.split(line.toString());String itemID = tokens[0];Double result = Double.parseDouble(tokens[1]);if(map.containsKey(itemID)){map.put(itemID, Double.parseDouble(map.get(itemID).toString()) + result);//矩阵乘法求和计算}else{map.put(itemID, result);}}Iterator iter = map.keySet().iterator();while (iter.hasNext()) {String itemID = (String) iter.next();double score = (double) map.get(itemID);Text v = new Text(itemID + "," + score);context.write(key, v);}}}public static void run(Map<String, String> path) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {// TODO Auto-generated method stubString input = path.get("hdfs_step4_updata2_input");String output = path.get("hdfs_step4_updata2_output");hdfsGYT hdfs = new hdfsGYT();hdfs.rmr(output);Job job = new Job(new Configuration(), "Step4_Updata2");job.setJarByClass(Step4_Updata2.class);FileInputFormat.addInputPath(job, new Path(input));FileOutputFormat.setOutputPath(job, new Path(output));//设置map和reduce类job.setMapperClass(Step4_Updata2_Map.class);job.setReducerClass(Step4_Updata_Reduce.class);//设置Map输出job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);//设置Reduce输出job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);//设置文件输入输出job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);job.waitForCompletion(true);}}

最终的运行结果截图：第一列为用户id，第二列为bookid，第三列为喜欢程度，由于数据集的关系，这里计算结果较大，在实际环境中，往往需要量化处理

六、推荐系统评估

1：用户满意度

      用户满意度是无法通过离线实验得到的，只能通过用户调查或者在线实验得到。      用户调查获得用户满意度主要是通过调查问卷的形式，用户对推荐系统的满意度分为不同的层次，GroupLens曾经做过一个论文推荐系统的调查问卷，该问卷的调查问题是请问下面哪句话最能描述你看到推荐结果后的感受     (1)推荐的论文都是我想看的     (2)推荐的论文很多我都看过了，确实是符合我兴趣的不错的论文     (3)推荐的论文和我的研究兴趣是相关的，但是我并不喜欢     (4)不知道为什么会推荐这些论文，它们和我的兴趣没有关系      由此可以看出，这个问卷不是简单的询问用户对结果的满意度，而是从不同的侧面询问用户对结果的不同感受。      在线实验，主要是记录一些用户的行为得到，比如说电子商务网站，用户买了推荐的商品，就表示它们在一定程度上喜欢。

2：预测准确度

     评分预测：针对用户给物品打分的推荐系统，比如豆瓣的评分      TopN推荐：网站在提供推荐服务时,一般是给用户一个个性化的推荐列表,这种推荐叫做 TopN 推荐，TopN 推荐的预测准确率一般通过准确率( precision ) / 召回率( recall )度量。

3：覆盖率

      覆盖率( coverage )描述一个推荐系统对物品长尾的发掘能力。覆盖率有不同的定义方法,最简单的定义为推荐系统能够推荐出来的物品占总物品集合的比例。假设系统的用户集合为 U ,推荐系统给每个用户推荐一个长度为N 的物品列表 R(u) 。那么推荐系统的覆盖率可以通过下面的公式计算:                                                                            从上面的定义可以看到,覆盖率是一个内容提供商会关心的指标。以图书推荐为例,出版社可能会很关心他们的书有没有被推荐给用户。覆盖率为 100% 的推荐系统可以将每个物品都推荐给至少一个用户。此外,从上面的定义也可以看到,热门排行榜的推荐覆盖率是很低的,它只会推荐那些热门的物品,这些物品在总物品中占的比例很小。一个好的推荐系统不仅需要有比较高的用户满意度,也要有较高的覆盖率。      在信息论和经济学中有两个著名的指标可以用来定义覆盖率。第一个是信息熵：                 第二个指标是基尼系数( Gini Index ):

4：多样性

      为了满足用户广泛的兴趣,推荐列表需要能够覆盖用户不同的兴趣领域,即推荐结果需要具有多样性。多样性描述了推荐列表中物品两两之间的不相似性。因此,多样性和相似性是对应的。假设s ( i , j )属于[0,1] 定义了物品 i 和 j 之间的相似度,那么用户 u 的推荐列表 R(u) 的多样性定义如下:                                                              而推荐系统的整体多样性可以定义为所有用户推荐列表多样性的平均值:                                                              从上面的定义可以看到,不同的物品相似度度量函数 s(i, j) 可以定义不同的多样性。如果用内容相似度描述物品间的相似度,我们就可以得到内容多样性函数,如果用协同过滤的相似度函数描述物品间的相似度,就可以得到协同过滤的多样性函数。

5：新颖性

      新颖的推荐是指给用户推荐那些他们以前没有听说过的物品。在一个网站中实现新颖性的最简单办法是,把那些用户之前在网站中对其有过行为的物品从推荐列表中过滤掉。

6：惊喜度

7：信任度

8：实时性

9：健壮性

10：商业目标