（转） 基于MapReduce的ItemBase推荐算法的共现矩阵实现（一）

 

转自：http://zengzhaozheng.blog.51cto.com/8219051/1557054

一、概述

这2个月研究根据用户标签情况对用户的相似度进行评估，其中涉及一些推荐算法知识，在这段时间研究了一遍《推荐算法实践》和《Mahout in action》，在这里主要是根据这两本书的一些思想和自己的一些理解对分布式基于ItemBase的推荐算法进行实现。其中分两部分，第一部分是根据共现矩阵的方式来简单的推算出用户的推荐项，第二部分则是通过传统的相似度矩阵的方法来实践ItemBase推荐算法。这篇blog主要记录第一部分的内容，并且利用MapReduce进行实现，下一篇blog则是记录第二部分的内容和实现。

二、算法原理

协同推荐算法，作为众多推荐算法中的一种已经被广泛的应用。其主要分为2种，第一种就是基于用户的协同过滤，第二种就是基于物品的协同过滤。

所谓的itemBase推荐算法简单直白的描述就是：用户A喜欢物品X1，用户B喜欢物品X2，如果X1和X2相似则，将A之前喜欢过的物品推荐给B，或者B之前喜欢过的物品推荐给A。这种算法是完全依赖于用户的历史喜欢物品的；所谓的UserBase推荐算法直白地说就是：用户A喜欢物品X1，用户B喜欢物品X2，如果用户A和用户B相似则将物品X1推荐给用户B，将物品X2推荐给用户A。简单的示意图：

至于选择哪种要看自己的实际情况，如果用户量比物品种类多得多那么就采用ItemBase的协同过滤推荐算法，如果是用户量比物品种类少的多则采用UserBase的协同顾虑推荐算，这样选择的一个原因是为了让物品的相似度矩阵或者用户相似度矩阵或者共现矩阵的规模最小化。

三、数据建模

基本的算法上面已经大概说了一下，对于算法来说，对数据建模使之运用在算法之上是重点也是难点。这小节主要根据自己相关项目的经验和《推荐引擎实践》的一些观点来讨论一些。分开2部分说，一是根据共现矩阵推荐、而是根据相似度算法进行推荐。

(1)共现矩阵方式：

第一步：转换成用户向量

1[102:0.1,103:0.2,104:0.3]：表示用户1喜欢的物品列表，以及他们对应的喜好评分。

2[101:0.1,102:0.7,105:0.9]：表示用户2喜欢的物品列表，以及他们对应的喜好评分。

3[102:0.1,103:0.7,104:0.2]：表示用户3喜欢的物品列表，以及他们对应的喜好评分。

第二步：计算共现矩阵

简单地说就是将同时喜欢物品x1和x2的用户数组成矩阵。

第三步：

生成用户对物品的评分矩阵

第四步：物品共现矩阵和用户对物品的评分矩阵相乘得到推荐结果

举个例子计算用户1的推荐列表过程：

用户1对物品101的总评分计算：

1*0+1*0.1+0*0.2+0*0.3+1*0=0.1

用户1对物品102的总评分计算：

1*0+3*0.1+1*0.2+2*0.3+2*0=1.1

用户1对物品103的总评分计算：

0*0+1*0.1+1*0.2+1*0.3+0*0=0.6

用户1对物品104的总评分计算：

0*0+2*0.1+1*0.2+2*0.3+1*0=1.0

用户1对物品105的总评分计算：

1*0+2*0.1+0*0.2+1*0.3+2*0=0.5

从而得到用户1的推荐列表为1[101:0.1,102:1.1,103:0.6,104:1.0,105:0.5]再经过排序得到最终推荐列表1[102:1.1,104:1.0,103:0.6,105:0.5,101:0.1]。

(2)通过计算机物品相似度方式计算用户的推荐向量。

通过计算机物品相似度方式计算用户的推荐向量和上面通过共现矩阵的方式差不多，就是将物品相似度矩阵代替掉共现矩阵和用户对物品的评分矩阵相乘，然后在计算推荐向量。

计算相似度矩阵：

在计算之前我们先了解一下物品相似度相关的计算方法。

对于计算物品相似度的算法有很多，要根据自己的数据模型进行选择。基于皮尔逊相关系数计算、欧几里德定理（实际上是算两点距离）、基于余弦相似度计算斯皮尔曼相关系数计算、基于谷本系数计算、基于对数似然比计算。其中谷本系数和对数似然比这两种方式主要是针对那些没有指名对物品喜欢度的数据模型进行相似度计算，也就是mahout中所指的Boolean数据模型。下面主要介绍2种，欧几里德和余弦相似度算法。

现在关键是怎么将现有数据转化成对应的空间向量模型使之适用这些定理，这是个关键点。下面我以欧几里德定理作为例子看看那如何建立模型：

第一步：将用户向量转化为物品向量

用户向量：

1[102:0.1,103:0.2,104:0.3]

2[101:0.1,102:0.7,105:0.9]

3[102:0.1,103:0.7,104:0.2]

转为为物品向量：

101[2:0.1]

102[1:0.1,2:0.7,3:0.1]

103[1:0.2,3:0.7]

104[1:0.3,3:0.2]

105[2:0.9]

第二步：

那么物品相似度计算为：

第三步：

最终得到物品相似度矩阵为：(这里省略掉没有意义的自关联相似度)

第四步：物品相似度矩阵和用户对物品的评分矩阵相乘得到推荐结果：

举个例子计算用户1的类似推荐列表过程：

用户1对物品101的总评分计算：

1*0+1*0.6186429+0*0.6964322+0*0.7277142+1*0.55555556=1.174198

用户1对物品102的总评分计算：

1*0.6186429+3*0+1*0.5188439+2*0.5764197+2*0.8032458=3.896818

用户1对物品103的总评分计算：

0*0.6964322+1*0.5188439+1*0+1*0.662294+0*0.463481=1.181138

用户1对物品104的总评分计算：

0*0.7277142+2*0.5764197+1*0.662294+2*0+1*0.5077338=2.322867

用户1对物品105的总评分计算：

1*0.55555556+2*0.8032458+0*0.463481+1*0.5077338=2.669780

四、共现矩阵方式的MapReduce实现

这里主要是利用MapReduce结合Mahout连的一些数据类型对共现矩阵方式的推荐方法进行实现,至于相似度矩阵方式进行推荐的在下一篇blog写。这里采用Boolean数据模型，即用户是没有对喜欢的物品进行初始打分的，我们在程序中默认都为1。

先看看整个MapReduce的数据流向图：

具体代码实现：HadoopUtil

  1. package com.util;
  3. import java.io.IOException;
  4. import java.util.Arrays;
  5. import java.util.Iterator;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.fs.FileSystem;
  8. import org.apache.hadoop.fs.Path;
  9. import org.apache.hadoop.fs.PathFilter;
  10. import org.apache.hadoop.io.Writable;
  11. import org.apache.hadoop.mapreduce.InputFormat;
  12. import org.apache.hadoop.mapreduce.Job;
  13. import org.apache.hadoop.mapreduce.JobContext;
  14. import org.apache.hadoop.mapreduce.Mapper;
  15. import org.apache.hadoop.mapreduce.OutputFormat;
  16. import org.apache.hadoop.mapreduce.Reducer;
  17. import org.apache.mahout.common.iterator.sequencefile.PathType;
  18. import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterator;
  19. import org.apache.mahout.common.iterator.sequencefile.SequenceFileValueIterator;
  20. import org.slf4j.Logger;
  21. import org.slf4j.LoggerFactory;
  23. publicfinalclassHadoopUtil{
  25. privatestaticfinalLogger log =LoggerFactory.getLogger(HadoopUtil.class);
  27. privateHadoopUtil(){}
  29. publicstaticJob prepareJob(String jobName,
  30. String[] inputPath,
  31. String outputPath,
  32. Class<?extendsInputFormat> inputFormat,
  33. Class<?extendsMapper> mapper,
  34. Class<?extendsWritable> mapperKey,
  35. Class<?extendsWritable> mapperValue,
  36. Class<?extendsOutputFormat> outputFormat,Configuration conf)throwsIOException{
  38. Job job =newJob(newConfiguration(conf));
  39. job.setJobName(jobName);
  40. Configuration jobConf = job.getConfiguration();
  42. if(mapper.equals(Mapper.class)){
  43. thrownewIllegalStateException("Can‘t figure out the user class jar file from mapper/reducer");
  44. }
  45. job.setJarByClass(mapper);
  47. job.setInputFormatClass(inputFormat);
  48. job.setInputFormatClass(inputFormat);
  49. StringBuilder inputPathsStringBuilder =newStringBuilder();
  50. for(String p : inputPath){
  51. inputPathsStringBuilder.append(",").append(p);
  52. }
  53. inputPathsStringBuilder.deleteCharAt(0);
  54. jobConf.set("mapred.input.dir", inputPathsStringBuilder.toString());
  56. job.setMapperClass(mapper);
  57. job.setMapOutputKeyClass(mapperKey);
  58. job.setMapOutputValueClass(mapperValue);
  59. job.setOutputKeyClass(mapperKey);
  60. job.setOutputValueClass(mapperValue);
  61. jobConf.setBoolean("mapred.compress.map.output",true);
  62. job.setNumReduceTasks(0);
  64. job.setOutputFormatClass(outputFormat);
  65. jobConf.set("mapred.output.dir", outputPath);
  67. return job;
  68. }
  70. publicstaticJob prepareJob(String jobName,
  71. String[] inputPath,
  72. String outputPath,
  73. Class<?extendsInputFormat> inputFormat,
  74. Class<?extendsMapper> mapper,
  75. Class<?extendsWritable> mapperKey,
  76. Class<?extendsWritable> mapperValue,
  77. Class<?extendsReducer> reducer,
  78. Class<?extendsWritable> reducerKey,
  79. Class<?extendsWritable> reducerValue,
  80. Class<?extendsOutputFormat> outputFormat,
  81. Configuration conf)throwsIOException{
  83. Job job =newJob(newConfiguration(conf));
  84. job.setJobName(jobName);
  85. Configuration jobConf = job.getConfiguration();
  87. if(reducer.equals(Reducer.class)){
  88. if(mapper.equals(Mapper.class)){
  89. thrownewIllegalStateException("Can‘t figure out the user class jar file from mapper/reducer");
  90. }
  91. job.setJarByClass(mapper);
  92. }else{
  93. job.setJarByClass(reducer);
  94. }
  96. job.setInputFormatClass(inputFormat);
  97. StringBuilder inputPathsStringBuilder =newStringBuilder();
  98. for(String p : inputPath){
  99. inputPathsStringBuilder.append(",").append(p);
  100. }
  101. inputPathsStringBuilder.deleteCharAt(0);
  102. jobConf.set("mapred.input.dir", inputPathsStringBuilder.toString());
  104. job.setMapperClass(mapper);
  105. if(mapperKey !=null){
  106. job.setMapOutputKeyClass(mapperKey);
  107. }
  108. if(mapperValue !=null){
  109. job.setMapOutputValueClass(mapperValue);
  110. }
  112. jobConf.setBoolean("mapred.compress.map.output",true);
  114. job.setReducerClass(reducer);
  115. job.setOutputKeyClass(reducerKey);
  116. job.setOutputValueClass(reducerValue);
  118. job.setOutputFormatClass(outputFormat);
  119. jobConf.set("mapred.output.dir", outputPath);
  121. return job;
  122. }
  127. publicstaticJob prepareJob(String jobName,String[] inputPath,
  128. String outputPath,Class<?extendsInputFormat> inputFormat,
  129. Class<?extendsMapper> mapper,
  130. Class<?extendsWritable> mapperKey,
  131. Class<?extendsWritable> mapperValue,
  132. Class<?extendsReducer> combiner,
  133. Class<?extendsReducer> reducer,
  134. Class<?extendsWritable> reducerKey,
  135. Class<?extendsWritable> reducerValue,
  136. Class<?extendsOutputFormat> outputFormat,Configuration conf)
  137. throwsIOException{
  139. Job job =newJob(newConfiguration(conf));
  140. job.setJobName(jobName);
  141. Configuration jobConf = job.getConfiguration();
  143. if(reducer.equals(Reducer.class)){
  144. if(mapper.equals(Mapper.class)){
  145. thrownewIllegalStateException(
  146. "Can‘t figure out the user class jar file from mapper/reducer");
  147. }
  148. job.setJarByClass(mapper);
  149. }else{
  150. job.setJarByClass(reducer);
  151. }
  153. job.setInputFormatClass(inputFormat);
  154. StringBuilder inputPathsStringBuilder =newStringBuilder();
  155. for(String p : inputPath){
  156. inputPathsStringBuilder.append(",").append(p);
  157. }
  158. inputPathsStringBuilder.deleteCharAt(0);
  159. jobConf.set("mapred.input.dir", inputPathsStringBuilder.toString());
  161. job.setMapperClass(mapper);
  162. if(mapperKey !=null){
  163. job.setMapOutputKeyClass(mapperKey);
  164. }
  165. if(mapperValue !=null){
  166. job.setMapOutputValueClass(mapperValue);
  167. }
  169. jobConf.setBoolean("mapred.compress.map.output",true);
  171. job.setCombinerClass(combiner);
  173. job.setReducerClass(reducer);
  174. job.setOutputKeyClass(reducerKey);
  175. job.setOutputValueClass(reducerValue);
  177. job.setOutputFormatClass(outputFormat);
  178. jobConf.set("mapred.output.dir", outputPath);
  180. return job;
  181. }
  183. publicstaticString getCustomJobName(String className,JobContext job,
  184. Class<?extendsMapper> mapper,
  185. Class<?extendsReducer> reducer){
  186. StringBuilder name =newStringBuilder(100);
  187. String customJobName = job.getJobName();
  188. if(customJobName ==null|| customJobName.trim().isEmpty()){
  189. name.append(className);
  190. }else{
  191. name.append(customJobName);
  192. }
  193. name.append(‘-‘).append(mapper.getSimpleName());
  194. name.append(‘-‘).append(reducer.getSimpleName());
  195. return name.toString();
  196. }
  199. publicstaticvoiddelete(Configuration conf,Iterable<Path> paths)throwsIOException{
  200. if(conf ==null){
  201. conf =newConfiguration();
  202. }
  203. for(Path path : paths){
  204. FileSystem fs = path.getFileSystem(conf);
  205. if(fs.exists(path)){
  206. log.info("Deleting {}", path);
  207. fs.delete(path,true);
  208. }
  209. }
  210. }
  212. publicstaticvoiddelete(Configuration conf,Path... paths)throwsIOException{
  213. delete(conf,Arrays.asList(paths));
  214. }
  216. publicstaticlong countRecords(Path path,Configuration conf)throwsIOException{
  217. long count =0;
  218. Iterator<?> iterator =newSequenceFileValueIterator<Writable>(path,true, conf);
  219. while(iterator.hasNext()){
  220. iterator.next();
  221. count++;
  222. }
  223. return count;
  224. }
  226. publicstaticlong countRecords(Path path,PathType pt,PathFilter filter,Configuration conf)throwsIOException{
  227. long count =0;
  228. Iterator<?> iterator =newSequenceFileDirValueIterator<Writable>(path, pt, filter,null,true, conf);
  229. while(iterator.hasNext()){
  230. iterator.next();
  231. count++;
  232. }
  233. return count;
  234. }
  235. }

先看看写的工具类：

第一步：处理原始输入数据

处理原始数据的SourceDataToItemPrefsJob作业的mapper：SourceDataToItemPrefsMapper

  1. package com.mapper;
  3. import java.io.IOException;
  4. import java.util.regex.Matcher;
  5. import java.util.regex.Pattern;
  7. import org.apache.hadoop.io.LongWritable;
  8. import org.apache.hadoop.io.Text;
  9. import org.apache.hadoop.mapreduce.Mapper;
  10. import org.apache.mahout.math.VarLongWritable;
  13. /**
  14. * mapper输入格式：userID:itemID1 itemID2 itemID3....
  15. * mapper输出格式:<userID,itemID>
  16. * @author 曾昭正
  17. */
  18. publicclassSourceDataToItemPrefsMapperextendsMapper<LongWritable,Text,VarLongWritable,VarLongWritable>{
  19. //private static final Logger logger = LoggerFactory.getLogger(SourceDataToItemPrefsMapper.class);
  20. privatestaticfinalPattern NUMBERS =Pattern.compile("(\\d+)");
  21. privateString line =null;
  23. @Override
  24. protectedvoid map(LongWritable key,Text value,Context context)
  25. throwsIOException,InterruptedException{
  26. line = value.toString();
  27. if(line ==null)return;
  28. // logger.info("line:"+line);
  29. Matcher matcher = NUMBERS.matcher(line);
  30. matcher.find();//寻找第一个分组，即userID
  31. VarLongWritable userID =newVarLongWritable(Long.parseLong(matcher.group()));//这个类型是在mahout中独立进行封装的
  32. VarLongWritable itemID =newVarLongWritable();
  33. while(matcher.find()){
  34. itemID.set(Long.parseLong(matcher.group()));
  35. // logger.info(userID + " " + itemID);
  36. context.write(userID, itemID);
  37. }
  38. }
  39. }

处理原始数据的SourceDataToItemPrefsJob作业的reducer：SourceDataToItemPrefsMapper

  1. package com.reducer;
  3. import java.io.IOException;
  5. import org.apache.hadoop.mapreduce.Reducer;
  6. import org.apache.mahout.math.RandomAccessSparseVector;
  7. import org.apache.mahout.math.VarLongWritable;
  8. import org.apache.mahout.math.Vector;
  9. import org.apache.mahout.math.VectorWritable;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  13. /**
  14. * reducer输入：<userID,Iterable<itemID>>
  15. * reducer输出:<userID,VecotrWriable<index=itemID,valuce=pres>....>
  16. * @author 曾昭正
  17. */
  18. publicclassSourceDataToUserVectorReducerextendsReducer<VarLongWritable,VarLongWritable,VarLongWritable,VectorWritable>{
  19. privatestaticfinalLogger logger =LoggerFactory.getLogger(SourceDataToUserVectorReducer.class);
  20. @Override
  21. protectedvoid reduce(VarLongWritable userID,Iterable<VarLongWritable> itemPrefs,Context context)
  22. throwsIOException,InterruptedException{
  23. /**
  24. * DenseVector，它的实现就是一个浮点数数组，对向量里所有域都进行存储，适合用于存储密集向量。
  25. RandomAccessSparseVector 基于浮点数的 HashMap 实现的，key 是整形 (int) 类型，value 是浮点数 (double) 类型，它只存储向量中不为空的值，并提供随机访问。
  26. SequentialAccessVector 实现为整形 (int) 类型和浮点数 (double) 类型的并行数组，它也只存储向量中不为空的值，但只提供顺序访问。
  27. 用户可以根据自己算法的需求选择合适的向量实现类，如果算法需要很多随机访问，应该选择 DenseVector 或者 RandomAccessSparseVector，如果大部分都是顺序访问，SequentialAccessVector 的效果应该更好。
  28. 介绍了向量的实现，下面我们看看如何将现有的数据建模成向量，术语就是“如何对数据进行向量化”，以便采用 Mahout 的各种高效的聚类算法。
  29. */
  30. Vector userVector =newRandomAccessSparseVector(Integer.MAX_VALUE,100);
  31. for(VarLongWritable itemPref : itemPrefs){
  32. userVector.set((int)itemPref.get(),1.0f);//RandomAccessSparseVector.set(index,value),用户偏好类型为boolean类型，将偏好值默认都为1.0f
  33. }
  34. logger.info(userID+" "+newVectorWritable(userVector));
  35. context.write(userID,newVectorWritable(userVector));
  36. }
  37. }

第二步：将SourceDataToItemPrefsJob作业的reduce输出结果组合成共现矩阵

UserVectorToCooccurrenceJob作业的mapper：UserVectorToCooccurrenceMapper

  1. package com.mapper;
  3. import java.io.IOException;
  4. import java.util.Iterator;
  6. import org.apache.hadoop.io.IntWritable;
  7. import org.apache.hadoop.mapreduce.Mapper;
  8. import org.apache.mahout.math.VarLongWritable;
  9. import org.apache.mahout.math.Vector;
  10. import org.apache.mahout.math.VectorWritable;
  12. /**
  13. * mapper输入：<userID,VecotrWriable<index=itemID,valuce=pres>....>
  14. * mapper输出:<itemID,itemID>(共现物品id对)
  15. * @author 曾昭正
  16. */
  17. publicclassUserVectorToCooccurrenceMapperextendsMapper<VarLongWritable,VectorWritable,IntWritable,IntWritable>{
  18. @Override
  19. protectedvoid map(VarLongWritable userID,VectorWritable userVector,Context context)
  20. throwsIOException,InterruptedException{
  21. Iterator<Vector.Element> it = userVector.get().nonZeroes().iterator();//过滤掉非空元素
  22. while(it.hasNext()){
  23. int index1 = it.next().index();
  24. Iterator<Vector.Element> it2 = userVector.get().nonZeroes().iterator();
  25. while(it2.hasNext()){
  26. int index2 = it2.next().index();
  27. context.write(newIntWritable(index1),newIntWritable(index2));
  28. }
  29. }
  31. }
  32. }

UserVectorToCooccurrenceJob作业的reducer：UserVectorToCoocurrenceReducer

  1. package com.reducer;
  3. import java.io.IOException;
  5. import org.apache.hadoop.io.IntWritable;
  6. import org.apache.hadoop.mapreduce.Reducer;
  7. import org.apache.mahout.cf.taste.hadoop.item.VectorOrPrefWritable;
  8. import org.apache.mahout.math.RandomAccessSparseVector;
  9. import org.apache.mahout.math.Vector;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  12. /**
  13. * reducer输入:<itemID,Iterable<itemIDs>>
  14. * reducer输出:<mainItemID,Vector<coocItemID,coocTime(共现次数)>....>
  15. * @author 曾昭正
  16. */
  17. publicclassUserVectorToCoocurrenceReducerextendsReducer<IntWritable,IntWritable,IntWritable,VectorOrPrefWritable>{
  18. privatestaticfinalLogger logger =LoggerFactory.getLogger(UserVectorToCoocurrenceReducer.class);
  19. @Override
  20. protectedvoid reduce(IntWritable mainItemID,Iterable<IntWritable> coocItemIDs,Context context)
  21. throwsIOException,InterruptedException{
  22. Vector coocItemIDVectorRow =newRandomAccessSparseVector(Integer.MAX_VALUE,100);
  23. for(IntWritable coocItem : coocItemIDs){
  24. int itemCoocTime = coocItem.get();
  25. coocItemIDVectorRow.set(itemCoocTime,coocItemIDVectorRow.get(itemCoocTime)+1.0);//将共现次数累加
  26. }
  27. logger.info(mainItemID +" "+newVectorOrPrefWritable(coocItemIDVectorRow));
  28. context.write(mainItemID,newVectorOrPrefWritable(coocItemIDVectorRow));//记录mainItemID的完整共现关系
  29. }
  30. }

第三步：将SourceDataToItemPrefsJob作业的reduce输出结果进行分割

userVecotrSplitJob作业的mapper：UserVecotrSplitMapper

  1. package com.mapper;
  3. import java.io.IOException;
  4. import java.util.Iterator;
  6. import org.apache.hadoop.io.IntWritable;
  7. import org.apache.hadoop.mapreduce.Mapper;
  8. import org.apache.mahout.cf.taste.hadoop.item.VectorOrPrefWritable;
  9. import org.apache.mahout.math.VarLongWritable;
  10. import org.apache.mahout.math.Vector;
  11. import org.apache.mahout.math.Vector.Element;
  12. import org.apache.mahout.math.VectorWritable;
  13. import org.slf4j.Logger;
  14. import org.slf4j.LoggerFactory;
  17. /**
  18. * 将用户向量分割，以便和物品的共现向量进行合并
  19. * mapper输入:<userID,Vector<itemIDIndex,preferenceValuce>....>
  20. * reducer输出:<itemID,Vecotor<userID,preferenceValuce>....>
  21. * @author 曾昭正
  22. */
  23. publicclassUserVecotrSplitMapperextendsMapper<VarLongWritable,VectorWritable,IntWritable,VectorOrPrefWritable>{
  24. privatestaticfinalLogger logger =LoggerFactory.getLogger(UserVecotrSplitMapper.class);
  25. @Override
  26. protectedvoid map(VarLongWritable userIDWritable,VectorWritable value,Context context)
  27. throwsIOException,InterruptedException{
  28. IntWritable itemIDIndex =newIntWritable();
  29. long userID = userIDWritable.get();
  30. Vector userVector = value.get();
  31. Iterator<Element> it = userVector.nonZeroes().iterator();//只取非空用户向量
  32. while(it.hasNext()){
  33. Element e = it.next();
  34. int itemID = e.index();
  35. itemIDIndex.set(itemID);
  36. float preferenceValuce =(float) e.get();
  37. logger.info(itemIDIndex +" "+newVectorOrPrefWritable(userID,preferenceValuce));
  38. context.write(itemIDIndex,newVectorOrPrefWritable(userID,preferenceValuce));
  39. }
  41. }
  42. }

第四步：将userVecotrSplitJob和UserVectorToCooccurrenceJob作业的输出结果合并

combineUserVectorAndCoocMatrixJob作业的mapper：CombineUserVectorAndCoocMatrixMapper

  1. package com.mapper;
  3. import java.io.IOException;
  5. import org.apache.hadoop.io.IntWritable;
  6. import org.apache.hadoop.mapreduce.Mapper;
  7. import org.apache.mahout.cf.taste.hadoop.item.VectorOrPrefWritable;
  9. /**
  10. * 将共现矩阵和分割后的用户向量进行合并，以便计算部分的推荐向量
  11. * 这个mapper其实没有什么逻辑处理功能，只是将数据按照输入格式输出
  12. * 注意：这里的mapper输入为共现矩阵和分割后的用户向量计算过程中的共同输出的2个目录
  13. * mapper输入：<itemID,Vecotor<userID,preferenceValuce>> or <itemID,Vecotor<coocItemID,coocTimes>>
  14. * mapper输出:<itemID,Vecotor<userID,preferenceValuce>/Vecotor<coocItemID,coocTimes>>
  15. * @author 曾昭正
  16. */
  17. publicclassCombineUserVectorAndCoocMatrixMapperextendsMapper<IntWritable,VectorOrPrefWritable,IntWritable,VectorOrPrefWritable>{
  18. @Override
  19. protectedvoid map(IntWritable itemID,VectorOrPrefWritable value,Context context)
  20. throwsIOException,InterruptedException{
  21. context.write(itemID, value);
  22. }
  24. }

combineUserVectorAndCoocMatrixJob作业的CombineUserVectorAndCoocMatrixReducer

  1. package com.reducer;
  3. import java.io.IOException;
  4. import java.util.ArrayList;
  5. import java.util.Iterator;
  6. import java.util.List;
  8. import org.apache.hadoop.io.IntWritable;
  9. import org.apache.hadoop.mapreduce.Reducer;
  10. import org.apache.mahout.cf.taste.hadoop.item.VectorAndPrefsWritable;
  11. import org.apache.mahout.cf.taste.hadoop.item.VectorOrPrefWritable;
  12. import org.apache.mahout.math.Vector;
  13. import org.slf4j.Logger;
  14. import org.slf4j.LoggerFactory;
  16. /**
  17. * 将共现矩阵和分割后的用户向量进行合并，以便计算部分的推荐向量
  18. * @author 曾昭正
  19. */
  20. publicclassCombineUserVectorAndCoocMatrixReducerextendsReducer<IntWritable,VectorOrPrefWritable,IntWritable,VectorAndPrefsWritable>{
  21. privatestaticfinalLogger logger =LoggerFactory.getLogger(CombineUserVectorAndCoocMatrixReducer.class);
  22. @Override
  23. protectedvoid reduce(IntWritable itemID,Iterable<VectorOrPrefWritable> values,Context context)
  24. throwsIOException,InterruptedException{
  25. VectorAndPrefsWritable vectorAndPrefsWritable =newVectorAndPrefsWritable();
  26. List<Long> userIDs =newArrayList<Long>();
  27. List<Float> preferenceValues =newArrayList<Float>();
  28. Vector coocVector =null;
  29. Vector coocVectorTemp =null;
  30. Iterator<VectorOrPrefWritable> it = values.iterator();
  31. while(it.hasNext()){
  32. VectorOrPrefWritable e = it.next();
  33. coocVectorTemp = e.getVector();
  34. if(coocVectorTemp ==null){
  35. userIDs.add(e.getUserID());
  36. preferenceValues.add(e.getValue());
  37. }else{
  38. coocVector = coocVectorTemp;
  39. }
  40. }
  41. if(coocVector !=null){
  42. //这个需要注意，根据共现矩阵的计算reduce聚合之后，到了这个一个Reudce分组就有且只有一个vecotr(即共现矩阵的一列或者一行，这里行和列是一样的)了。
  43. vectorAndPrefsWritable.set(coocVector, userIDs, preferenceValues);
  44. logger.info(itemID+" "+vectorAndPrefsWritable);
  45. context.write(itemID, vectorAndPrefsWritable);
  46. }
  47. }
  48. }

第五步：将combineUserVectorAndCoocMatrixJob作业的输出结果生成推荐列表

caclPartialRecomUserVectorJob作业的mapper：CaclPartialRecomUserVectorMapper

  1. package com.mapper;
  3. import java.io.IOException;
  4. import java.util.List;
  6. import org.apache.hadoop.io.IntWritable;
  7. import org.apache.hadoop.mapreduce.Mapper;
  8. import org.apache.mahout.cf.taste.hadoop.item.VectorAndPrefsWritable;
  9. import org.apache.mahout.math.VarLongWritable;
  10. import org.apache.mahout.math.Vector;
  11. import org.apache.mahout.math.VectorWritable;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  15. /**
  16. * 计算部分用户推荐向量
  17. * @author 曾昭正
  18. */
  19. publicclassCaclPartialRecomUserVectorMapperextendsMapper<IntWritable,VectorAndPrefsWritable,VarLongWritable,VectorWritable>{
  20. privatestaticfinalLogger logger =LoggerFactory.getLogger(CaclPartialRecomUserVectorMapper.class);
  21. @Override
  22. protectedvoid map(IntWritable itemID,VectorAndPrefsWritable values,Context context)
  23. throwsIOException,InterruptedException{
  24. Vector coocVectorColumn = values.getVector();
  25. List<Long> userIDs = values.getUserIDs();
  26. List<Float> preferenceValues = values.getValues();
  27. for(int i =0; i< userIDs.size(); i++){
  28. long userID = userIDs.get(i);
  29. float preferenceValue = preferenceValues.get(i);
  30. logger.info("userID:"+ userID);
  31. logger.info("preferenceValue:"+preferenceValue);
  32. //将共现矩阵中userID对应的列相乘，算出部分用户对应的推荐列表分数
  33. Vector preferenceParScores = coocVectorColumn.times(preferenceValue);
  34. context.write(newVarLongWritable(userID),newVectorWritable(preferenceParScores));
  35. }
  36. }
  37. }

caclPartialRecomUserVectorJob作业的combiner：ParRecomUserVectorCombiner

  1. package com.reducer;
  3. import java.io.IOException;
  5. import org.apache.hadoop.mapreduce.Reducer;
  6. import org.apache.mahout.math.VarLongWritable;
  7. import org.apache.mahout.math.Vector;
  8. import org.apache.mahout.math.VectorWritable;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. /**
  12. * 将计算部分用户推荐向量的结果进行合并，将userID对应的贡现向量的分值进行相加(注意：这个只是将一个map的输出进行合并，所以这个也是只部分结果)
  13. * @author 曾昭正
  14. */
  15. publicclassParRecomUserVectorCombinerextendsReducer<VarLongWritable,VectorWritable,VarLongWritable,VectorWritable>{
  16. privatestaticfinalLogger logger =LoggerFactory.getLogger(ParRecomUserVectorCombiner.class);
  17. @Override
  18. protectedvoid reduce(VarLongWritable userID,Iterable<VectorWritable> coocVectorColunms,Context context)
  19. throwsIOException,InterruptedException{
  21. Vector vectorColunms =null;
  23. for(VectorWritable coocVectorColunm : coocVectorColunms){
  24. vectorColunms = vectorColunms ==null? coocVectorColunm.get(): vectorColunms.plus(coocVectorColunm.get());
  25. }
  26. logger.info(userID +" "+newVectorWritable(vectorColunms));
  27. context.write(userID,newVectorWritable(vectorColunms));
  28. }
  29. }

caclPartialRecomUserVectorJob作业的reducer：MergeAndGenerateRecommendReducer

  1. package com.reducer;
  3. import java.io.IOException;
  4. import java.util.ArrayList;
  5. import java.util.Collections;
  6. import java.util.Iterator;
  7. import java.util.List;
  8. import java.util.PriorityQueue;
  9. import java.util.Queue;
  11. import org.apache.hadoop.mapreduce.Reducer;
  12. import org.apache.mahout.cf.taste.hadoop.RecommendedItemsWritable;
  13. import org.apache.mahout.cf.taste.impl.recommender.ByValueRecommendedItemComparator;
  14. import org.apache.mahout.cf.taste.impl.recommender.GenericRecommendedItem;
  15. import org.apache.mahout.cf.taste.recommender.RecommendedItem;
  16. import org.apache.mahout.math.VarLongWritable;
  17. import org.apache.mahout.math.Vector;
  18. import org.apache.mahout.math.Vector.Element;
  19. import org.apache.mahout.math.VectorWritable;
  20. import org.slf4j.Logger;
  21. import org.slf4j.LoggerFactory;
  23. /**
  24. * 合并所有已经评分的共现矩阵
  25. * @author 曾昭正
  26. */
  27. publicclassMergeAndGenerateRecommendReducerextendsReducer<VarLongWritable,VectorWritable,VarLongWritable,RecommendedItemsWritable>{
  28. privatestaticfinalLogger logger =LoggerFactory.getLogger(MergeAndGenerateRecommendReducer.class);
  29. privateint recommendationsPerUser;
  30. @Override
  31. protectedvoid setup(Context context)
  32. throwsIOException,InterruptedException{
  33. recommendationsPerUser = context.getConfiguration().getInt("recomandItems.recommendationsPerUser",5);
  34. }
  35. @Override
  36. protectedvoid reduce(VarLongWritable userID,Iterable<VectorWritable> cooVectorColumn,Context context)
  37. throwsIOException,InterruptedException{
  38. //分数求和合并
  39. Vector recommdVector =null;
  40. for(VectorWritable vector : cooVectorColumn){
  41. recommdVector = recommdVector ==null? vector.get(): recommdVector.plus(vector.get());
  42. }
  43. //对推荐向量进行排序，为每个UserID找出topM个推荐选项(默认找出5个)，此队列按照item对应的分数进行排序
  44. //注意下：PriorityQueue队列的头一定是最小的元素,另外这个队列容量增加1是为了为添加更大的新元素时使用的临时空间
  45. Queue<RecommendedItem> topItems =newPriorityQueue<RecommendedItem>(recommendationsPerUser+1,ByValueRecommendedItemComparator.getInstance());
  47. Iterator<Element> it = recommdVector.nonZeroes().iterator();
  48. while(it.hasNext()){
  49. Element e = it.next();
  50. int itemID = e.index();
  51. float preValue =(float) e.get();
  52. //当队列容量小于推荐个数，往队列中填item和分数
  53. if(topItems.size()< recommendationsPerUser){
  54. topItems.add(newGenericRecommendedItem(itemID, preValue));
  55. }
  56. //当前item对应的分数比队列中的item的最小分数大，则将队列头原始（即最小元素）弹出，并且将当前item：分数加入队列
  57. elseif(preValue > topItems.peek().getValue()){
  58. topItems.add(newGenericRecommendedItem(itemID, preValue));
  59. //弹出头元素（最小元素）
  60. topItems.poll();
  61. }
  62. }
  64. //重新调整队列的元素的顺序
  65. List<RecommendedItem> recommdations =newArrayList<RecommendedItem>(topItems.size());
  66. recommdations.addAll(topItems);//将队列中所有元素添加即将排序的集合
  67. Collections.sort(recommdations,ByValueRecommendedItemComparator.getInstance());//排序
  69. //输出推荐向量信息
  70. logger.info(userID+" "+newRecommendedItemsWritable(recommdations));
  71. context.write(userID,newRecommendedItemsWritable(recommdations));
  73. }
  74. }

第六步：组装各个作业关系

PackageRecomendJob

  1. package com.mapreduceMain;
  3. import java.io.IOException;
  4. import java.net.URI;
  6. import org.apache.hadoop.conf.Configuration;
  7. import org.apache.hadoop.conf.Configured;
  8. import org.apache.hadoop.fs.FileSystem;
  9. import org.apache.hadoop.fs.Path;
  10. import org.apache.hadoop.io.IntWritable;
  11. import org.apache.hadoop.mapreduce.Job;
  12. import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
  13. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
  14. import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
  15. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  16. import org.apache.hadoop.util.Tool;
  17. import org.apache.hadoop.util.ToolRunner;
  18. import org.apache.mahout.cf.taste.hadoop.RecommendedItemsWritable;
  19. import org.apache.mahout.cf.taste.hadoop.item.VectorAndPrefsWritable;
  20. import org.apache.mahout.cf.taste.hadoop.item.VectorOrPrefWritable;
  21. import org.apache.mahout.math.VarLongWritable;
  22. import org.apache.mahout.math.VectorWritable;
  23. import com.mapper.CaclPartialRecomUserVectorMapper;
  24. import com.mapper.CombineUserVectorAndCoocMatrixMapper;
  25. import com.mapper.UserVecotrSplitMapper;
  26. import com.mapper.UserVectorToCooccurrenceMapper;
  27. import com.mapper.SourceDataToItemPrefsMapper;
  28. import com.reducer.CombineUserVectorAndCoocMatrixReducer;
  29. import com.reducer.MergeAndGenerateRecommendReducer;
  30. import com.reducer.ParRecomUserVectorCombiner;
  31. import com.reducer.UserVectorToCoocurrenceReducer;
  32. import com.reducer.SourceDataToUserVectorReducer;
  33. import com.util.HadoopUtil;
  36. /**
  37. * 组装各个作业组件，完成推荐作业
  38. * @author 曾昭正
  39. */
  40. publicclassPackageRecomendJobextendsConfiguredimplementsTool{
  41. String[] dataSourceInputPath ={"/user/hadoop/z.zeng/distruteItemCF/dataSourceInput"};
  42. String[] uesrVectorOutput ={"/user/hadoop/z.zeng/distruteItemCF/uesrVectorOutput/"};
  43. String[] userVectorSpliltOutPut ={"/user/hadoop/z.zeng/distruteItemCF/userVectorSpliltOutPut"};
  44. String[] cooccurrenceMatrixOuptPath ={"/user/hadoop/z.zeng/distruteItemCF/CooccurrenceMatrixOuptPath"};
  45. String[] combineUserVectorAndCoocMatrixOutPutPath ={"/user/hadoop/z.zeng/distruteItemCF/combineUserVectorAndCoocMatrixOutPutPath"};
  46. String[] caclPartialRecomUserVectorOutPutPath ={"/user/hadoop/z.zeng/distruteItemCF/CaclPartialRecomUserVectorOutPutPath"};
  48. protectedvoid setup(Configuration configuration)
  49. throwsIOException,InterruptedException{
  50. FileSystem hdfs =FileSystem.get(URI.create("hdfs://cluster-master"), configuration);
  51. Path p1 =newPath(uesrVectorOutput[0]);
  52. Path p2 =newPath(userVectorSpliltOutPut[0]);
  53. Path p3 =newPath(cooccurrenceMatrixOuptPath[0]);
  54. Path p4 =newPath(combineUserVectorAndCoocMatrixOutPutPath[0]);
  55. Path p5 =newPath(caclPartialRecomUserVectorOutPutPath[0]);
  57. if(hdfs.exists(p1)){
  58. hdfs.delete(p1,true);
  59. }
  60. if(hdfs.exists(p2)){
  61. hdfs.delete(p2,true);
  62. }
  63. if(hdfs.exists(p3)){
  64. hdfs.delete(p3,true);
  65. }
  66. if(hdfs.exists(p4)){
  67. hdfs.delete(p4,true);
  68. }
  69. if(hdfs.exists(p5)){
  70. hdfs.delete(p5,true);
  71. }
  72. }
  73. @Override
  74. publicint run(String[] args)throwsException{
  75. Configuration conf=getConf();//获得配置文件对象
  76. setup(conf);
  77. // DistributedCache.addArchiveToClassPath(new Path("/user/hadoop/z.zeng/distruteItemCF/lib"), conf);
  78. //配置计算用户向量作业
  79. Job wikipediaToItemPrefsJob =HadoopUtil.prepareJob(
  80. "WikipediaToItemPrefsJob",
  81. dataSourceInputPath,
  82. uesrVectorOutput[0],
  83. TextInputFormat.class,
  84. SourceDataToItemPrefsMapper.class,
  85. VarLongWritable.class,
  86. VarLongWritable.class,
  87. SourceDataToUserVectorReducer.class,
  88. VarLongWritable.class,
  89. VectorWritable.class,
  90. SequenceFileOutputFormat.class,
  91. conf);
  92. //配置计算共现向量作业
  93. Job userVectorToCooccurrenceJob =HadoopUtil.prepareJob(
  94. "UserVectorToCooccurrenceJob",
  95. uesrVectorOutput,
  96. cooccurrenceMatrixOuptPath[0],
  97. SequenceFileInputFormat.class,
  98. UserVectorToCooccurrenceMapper.class,
  99. IntWritable.class,
  100. IntWritable.class,
  101. UserVectorToCoocurrenceReducer.class,
  102. IntWritable.class,
  103. VectorOrPrefWritable.class,
  104. SequenceFileOutputFormat.class,
  105. conf);
  106. //配置分割用户向量作业
  107. Job userVecotrSplitJob =HadoopUtil.prepareJob(
  108. "userVecotrSplitJob",
  109. uesrVectorOutput,
  110. userVectorSpliltOutPut[0],
  111. SequenceFileInputFormat.class,
  112. UserVecotrSplitMapper.class,
  113. IntWritable.class,
  114. VectorOrPrefWritable.class,
  115. SequenceFileOutputFormat.class,
  116. conf);
  117. //合并共现向量和分割之后的用户向量作业
  118. //这个主意要将分割用户向量和共现向量的输出结果一起作为输入
  119. String[] combineUserVectorAndCoocMatrixIutPutPath ={cooccurrenceMatrixOuptPath[0],userVectorSpliltOutPut[0]};
  120. Job combineUserVectorAndCoocMatrixJob =HadoopUtil.prepareJob(
  121. "combineUserVectorAndCoocMatrixJob",
  122. combineUserVectorAndCoocMatrixIutPutPath,
  123. combineUserVectorAndCoocMatrixOutPutPath[0],
  124. SequenceFileInputFormat.class,
  125. CombineUserVectorAndCoocMatrixMapper.class,
  126. IntWritable.class,
  127. VectorOrPrefWritable.class,
  128. CombineUserVectorAndCoocMatrixReducer.class,
  129. IntWritable.class,
  130. VectorAndPrefsWritable.class,
  131. SequenceFileOutputFormat.class,
  132. conf);
  133. //计算用户推荐向量
  134. Job caclPartialRecomUserVectorJob=HadoopUtil.prepareJob(
  135. "caclPartialRecomUserVectorJob",
  136. combineUserVectorAndCoocMatrixOutPutPath,
  137. caclPartialRecomUserVectorOutPutPath[0],
  138. SequenceFileInputFormat.class,
  139. CaclPartialRecomUserVectorMapper.class,
  140. VarLongWritable.class,
  141. VectorWritable.class,
  142. ParRecomUserVectorCombiner.class,//为map设置combiner减少网络IO
  143. MergeAndGenerateRecommendReducer.class,
  144. VarLongWritable.class,
  145. RecommendedItemsWritable.class,
  146. TextOutputFormat.class,
  147. conf);
  149. //串联各个job
  150. if(wikipediaToItemPrefsJob.waitForCompletion(true)){
  151. if(userVectorToCooccurrenceJob.waitForCompletion(true)){
  152. if(userVecotrSplitJob.waitForCompletion(true)){
  153. if(combineUserVectorAndCoocMatrixJob.waitForCompletion(true)){
  154. int rs = caclPartialRecomUserVectorJob.waitForCompletion(true)?1:0;
  155. return rs;
  156. }else{
  157. thrownewException("合并共现向量和分割之后的用户向量作业失败！！");
  158. }
  159. }else{
  160. thrownewException("分割用户向量作业失败！！");
  161. }
  162. }else{
  163. thrownewException("计算共现向量作业失败！！");
  164. }
  165. }else{
  166. thrownewException("计算用户向量作业失败！！");
  167. }
  168. }
  169. publicstaticvoid main(String[] args)throwsIOException,
  170. ClassNotFoundException,InterruptedException{
  171. try{
  172. int returnCode =ToolRunner.run(newPackageRecomendJob(),args);
  173. System.exit(returnCode);
  174. }catch(Exception e){
  175. }
  176. }
  178. }

五、总结

本blog主要说了下itemBase推荐算法的一些概念，以及如何多现有数据进行建模。其中对共现矩阵方式的推荐用MapReduce结合Mahout的内置数据类型进行了实现。写完这篇blog和对算法实现完毕后，发现Mapreduce编程虽然数据模型非常简单，只有2个过程：数据的分散与合并，但是在分散与合并的过程中可以使用自定义的各种数据组合类型使其能够完成很多复杂的功能。

参考文献：《Mahout in action》、《推荐引擎实践》

来自为知笔记(Wiz)

时间： 2024-08-05 06:28:54

（转） 基于MapReduce的ItemBase推荐算法的共现矩阵实现（一）的相关文章

Mahout in Action 学习---基于物品的分布式推荐算法（Wikipedia数据集）

文字总结自<Mahout in Action>中文版第六章的内容 1.1 数据集介绍 Wikipedia数据集:一篇文章到另外一篇文章的链接. 可以将文章看作是用户,将该文章指向的文章视为该源文章所喜欢的物品. 类型:单向布尔型偏好. 相似性评估算法:LogLikelihoodSimilarity 关于LogLikelihoodSimilarity具体算法思想见: 对数似然比相似度 - xidianycy - 博客频道 - CSDN.NET http://blog.csdn.net/u0143

基于用户的协同过滤推荐算法

什么是推荐算法 推荐算法最早在1992年就提出来了,但是火起来实际上是最近这些年的事情,因为互联网的爆发,有了更大的数据量可以供我们使用,推荐算法才有了很大的用武之地. 最开始,所以我们在网上找资料,都是进yahoo,然后分门别类的点进去,找到你想要的东西,这是一个人工过程,到后来,我们用google,直接搜索自 己需要的内容,这些都可以比较精准的找到你想要的东西,但是,如果我自己都不知道自己要找什么肿么办?最典型的例子就是,如果我打开豆瓣找电影,或者我去 买说,我实际上不知道我想要买什么或者看

基于特征的推荐算法【转】

http://in.sdo.com/?p=2779 推荐算法准确度度量公式: 其中,R(u)表示对用户推荐的N个物品,T(u)表示用户u在测试集上喜欢的物品集合. 集合相似度度量公式(N维向量的距离度量公式): Jaccard公式: 其中,N(u)表示用户u有过正反馈的物品集合. 余弦相似度公式: UserCF公式: 其中,S(u,k)表示和用户u兴趣最接近的K个用户集合:N(i)表示对物品i有过正反馈的用户集合:w(u,v)表示用户u和用户v的兴趣相似度:r(v,i)表示用户v对物品i的兴趣.

Mahout推荐算法API详解

前言 用Mahout来构建推荐系统,是一件既简单又困难的事情.简单是因为Mahout完整地封装了“协同过滤”算法,并实现了并行化,提供非常简单的API接口:困难是因为我们不了解算法细节,很难去根据业务的场景进行算法配置和调优. 本文将深入算法API去解释Mahout推荐算法底层的一些事. 1. Mahout推荐算法介绍 Mahoutt推荐算法,从数据处理能力上,可以划分为2类: 单机内存算法实现 基于Hadoop的分步式算法实现 1). 单机内存算法实现 单机内存算法实现:就是在单机下运行的算法

Mahout推荐算法API详解【一起学Mahout】

阅读导读: 1.mahout单机内存算法实现和分布式算法实现分别存在哪些问题? 2.算法评判标准有哪些? 3.什么会影响算法的评分? 1. Mahout推荐算法介绍 Mahout推荐算法,从数据处理能力上,可以划分为2类: 单机内存算法实现 基于Hadoop的分步式算法实现 1). 单机内存算法实现 单机内存算法实现:就是在单机下运行的算法,是由cf.taste项目实现的,像我们熟悉的UserCF,ItemCF都支持单机内存运行,并且参数可以灵活配置.单机算法的基本实例,请参考文章:用Maven

Mahout推荐算法API具体解释【一起学Mahout】

阅读导读: 1.mahout单机内存算法实现和分布式算法实现分别存在哪些问题? 2.算法评判标准有哪些? 3.什么会影响算法的评分? 1. Mahout推荐算法介绍 Mahout推荐算法,从数据处理能力上,能够划分为2类: 单机内存算法实现 基于Hadoop的分步式算法实现 1). 单机内存算法实现 单机内存算法实现:就是在单机下执行的算法,是由cf.taste项目实现的,像我们熟悉的UserCF,ItemCF都支持单机内存执行.而且參数能够灵活配置.单机算法的基本实例.请參考文章:用Maven

[转]Mahout推荐算法API详解

Mahout推荐算法API详解 Hadoop家族系列文章,主要介绍Hadoop家族产品,常用的项目包括Hadoop, Hive, Pig, HBase, Sqoop, Mahout, Zookeeper, Avro, Ambari, Chukwa,新增加的项目包括,YARN, Hcatalog, Oozie, Cassandra, Hama, Whirr, Flume, Bigtop, Crunch, Hue等. 从2011年开始,中国进入大数据风起云涌的时代,以Hadoop为代表的家族软件,占

Mahout推荐算法之ItemBased

Mahout推荐之ItemBased 一.   算法原理 (一)    基本原理 如下图评分矩阵所示:行为user,列为item. 图(1) 该算法的原理: 1.  计算Item之间的相似度. 2.  对用户U做推荐 公式(一) Map tmp ; Map tmp1 ; for(item a  in userRatedItems){ rate  =userforItemRate(a) ListsimItem =getSimItem(a); For(Jin simItem){ Item b =j;

Mahout推荐算法基础

转载自(http://www.geek521.com/?p=1423) Mahout推荐算法分为以下几大类 GenericUserBasedRecommender 算法: 1.基于用户的相似度 2.相近的用户定义与数量 特点: 1.易于理解 2.用户数较少时计算速度快 GenericItemBasedRecommender 算法: 1.基于item的相似度 特点: 1.item较少时就算速度更快 2.当item的外部概念易于理解和获得是非常有用 SlopeOneRecommender(itemB