mahout基于hadoop的CF代码分析

mahout的taste框架是协同过滤算法的实现。它支持DataModel，如文件、数据库、NoSQL存储等，也支持hadoop的MapReduce。这里主要分析的基于MR的实现。

基于MR的CF实现主要流程就在org.apache.mahout.cf.taste.hadoop.item.RecommenderJob类中（注意mahout有两个RecommendJob，要看清楚是哪一个包）。这个类的run方法就包含了所有的步骤。从上到下，完整的其实有10步（中间计算item相似度其实拆分成了3个job，我们也当做是一个phase吧），也就是说，如果指定了所有的必要参数，运行一次item-based CF算法，会执行12个JOB，当然有的步骤是可以忽略的，下面会讲。以下就是详细的每一步骤的分析：

phase1: itemIDIndex

这步主要是将itemId转成一个int。这里设计上其实有点小问题，如果item的数量非常多，比如超过int的最大值，那就有可能会出现重合了。所以用long其实更为合适。

input:用户评分文件（这也是我们最原始的输入了），格式一般为：userId \t itemId \t score。注意输入必须是textfile的。可能是为了方便测试吧，mahout的很多包默认输入都是textfile格式的。

map:(index, itemId)

reduce: (index, itemId)

phase2: toUserVector

input:用户评分文件

param: --userBooleanData如果这个参数为true，则会忽略评分列，对于买或不买这类数据有时需要指这定这个值。

map: (userId, itemId,pref)

reduce: 以用户为key，输出成向量形式==> (userId, VectorWritable<itemId, pref>)

phase3: countUser，计算用户数

map: (userId)

reduce: 输出总用户数count

phase4: maybePruneAndTranspose

input: phase2的输出：userVector

param: --maxCooccurrences

map: (userId,Vector<itemId, pref>) ==>(itemId,DistributedRowMatrix<userId,pref>)，注意如果指定了—maxCooccurrences参数，这里会有裁剪，每个userId最多对maxCooccurrences的itemId打分

这里的DistributedRowMatrix，分布式行矩阵：行：itemId,列：userId

reduce: (itemId, VectorWritable<userId,pref>)

phase5: RowSimilarityJob

这一步比较关键，计算item相似度，它拆分成了三个JOB。

param: --numberOfColumns, --similarityClassname,--maxSimilaritiesPerRow(默认:100)

job1:weight

input:phase4的输出

map: (itemId, VectorWritable <userId, pref>) ==>(userId, WeightedOccurrence<itemId, pref, weight>)

这里的weight，对于欧氏向量距离，或者Pearson距离等，均为Double.NaN，即无效。在LoglikelihoodVectorSimilarity中有用到weight的值。

reduce:(userId, WeightedOccurrenceArray<itemId, pref, weight>)

job2:pairwise similarity *item相似度计算*

map: 对同一用户的所有item-rating，输出两两item之间的关系 ==>(WeightedRowPair<itemA, itemB, weightA, weightB>, coocurrence<userId,valueA, valueB>) (同上，这里的权重weightA,B对于欧氏距离等可以忽略)

reduce: 在这端，以<itemA,itemB>为key聚合了来自不同map的所有用户的打分，最后输出itemA和B的对称相似度（即以itemA为key或以itemB为key）==>(SimilarityMatrixEntryKey<itemA,similarity>, MatrixEntryWritable<WeightedRowPair<itemA, itemB,weightA, weightB>>) ，(SimilarityMatrixEntryKey<itemB,similarity>, MatrixEntryWritable<WeightedRowPair<itemB, itemA,weightB, weightA>>)

job3:entries2vectors *汇总item的相似items*

param: --maxSimilaritiesPerRow

map: (itemA, itemB, similarity) & (itemB,itemA, similarity) 这里在group的时候按相似度降序做了排序，如果有--maxSimilaritiesPerRow参数，则会做裁剪。

reduce: (itemA, VectorWritable <item,similarity>)

至此，item相似度计算完毕。

phase6: prePartialMultiply1

input: phase5的最后输出(即item相似度)

map: 直接输出item对应的相似items，这里用VectorOrPrefWritable做了封装，表明有可能是相似度向量，也有可能是对item的打分，并且对item为自己的，将相似度设为Double.NaN，以过滤自身。==>(itemId,VectorOrPrefWritable<item, similarity>)

reduce: IdentityReducer

phase7: prePartialMultiply2

input: phase2的输出userVectors

map: 输出：(itemId, VectorOrPrefWritable<userId, pref>)

这里默认考虑用户对10个item的评分，可以通过maxPrefsPerUserConsidered参数调整。

如果指定了usersFile，则在setup时会把所有的userId读入内存，用于过滤。如果map输入数据的userID不在usersFile中，则会被忽略。注意，这是mahout的设计bug，对于比较大的数据集，很有可能造成OOM（事实上在我的测试程序中已经出现OOM了…），这种bug下面还会出现。输出的是用户的评分，同phase6的VectorOrPrefWritable的封装。

reduce: IdentityReducer

phase8: partialMultiply

input: 6和7的输出：prePartialMultiply1, prePartialMultiply2

map: Identity。由于6和7的输出的key均为itemId，因而在reduce端同一item的相似item以及对应的用户评分会聚合到一起。

reduce:(itemId, VectorAndPrefsWritable<similarityMatrix, List<userId>,List<pref>>) 没做特殊处理，直接合在一起，输出相似度矩阵，所有的userId及对item的打分。

phase9: itemFiltering

将过滤文件输出成<userId, itemId>。如果指定了--filterFile参数，则在最后的聚合推荐中会过滤userId对应的items。这一步在实际中多数是可以忽略的，只要不指定这个参数即可。

phase10: aggregateAndRecommend

map: 对每个用户，输出对当前item的评分，以及与当前item的所有相似items==>(userId, PrefAndSimilarityColumnWritable<pref,vector<item, similarity>>)

reduce: 聚合了这个用户所有的评分历史，以及相似items，计算对该用户的推荐结果 ==> (userId, List<itemId>)。

注意在reduce的setup中，会将phase1产生的所有itemId到index的映射读入内存，这里只要Item数据集稍大，就会OOM。这是比较严重的设计bug。

事实上，如果item是正规的整数，而不是guid之类的，phase1和这一步的读入内存是完全可以略掉的。这样的话就完全可以在企业级的数据集上使用（我的测试数据集是15亿+的user-item-rating，1.5亿+的用户，在最后这一步挂掉了，前面所有phase都能跑成功）。

至此，已经形成了推荐结果，CF完成。

以上的所有步骤中，phase5的计算item相似度是最慢的（这个其实也很直觉）。

总体上，mahout有几处比较严重的设计问题，这么设计的主要原因，是很多内部的Writable，如VectorWritable，它们的Index都是int型的，所以如果要改，就会牵一发而动全身。

对于超大规模数据集，建议还是裁剪一下吧，distinct item过亿一般是会挂的，最好能够控制在千万的级别内。

mahout基于hadoop的CF代码分析相关推荐

mahout基于Hadoop的CF代码分析(转）
来自:http://www.codesky.net/article/201206/171862.html mahout的taste框架是协同过滤算法的实现.它支持DataModel,如文件.数据库.N ...
（三.1）基于MicroBlaze的PowerlinkC代码分析
本文简单介绍ISE平台下的MicroBlaze的大致框架,然后分析开源的C代码在平台中的是使用以及通信原理的分析与笔者本身的理解.本文不介绍本次使用的是ISE14.7搭建平台,使用xilinx_op ...
python大数据处理mapreduce_使用python构建基于hadoop的mapreduce日志分析平台
流量比较大的日志要是直接写入Hadoop对Namenode负载过大,所以入库前合并,可以把各个节点的日志凑并成一个文件写入HDFS. 根据情况定期合成,写入到hdfs里面. 咱们看看日志的大小,200 ...
基于Hadoop的数据分析平台搭建
企业发展到一定规模都会搭建单独的BI平台来做数据分析,即OLAP(联机分析处理),一般都是基于数据库技术来构建,基本都是单机产品.除了业务数据的相关分析外,互联网企业还会对用户行为进行分析,进一步挖掘 ...
基于物品的协同过滤mysql_使用Mahout基于用户和物品相似度进行协同过滤推荐内容...
基于用户User协同过滤 package mahout; import java.io.File; import java.io.IOException; import java.util.List; ...
clang-tidy——静态代码分析框架
clang-tidy--静态代码分析框架 clang-tidy介绍 clang-tidy是一个基于clang的静态代码分析框架,支持C++/C/Objective-C. 它是一个功能更强大的lint工 ...
java hadoop mahout_一个基于Mahout与hadoop的聚类搭建
mahout是基于Hadoop的数据挖掘工具,因为有了hadoop,所以进行海量数据的挖掘工作显得更为简单.但是因为算法需要支持M/R,所以不是所有常用的数据挖掘算法都会支持.这篇文章会告诉你,如何使 ...
【Hadoop】基于Hadoop/Hbase/Hive的小型离线服务器日志分析应用
##1.项目简介本项目主要设计一个基于Hadoop的日志分析系统.其中的日志数据主要来自于某系统开发测试期间的服务器访问日志数据(Tomcat),通过对这些日志数据利用正则表达式等技术手段进行处理, ...
基于windows PE文件的恶意代码分析；使用SystemInternal工具与内核调试器研究windows用户空间与内核空间...
基于windows PE文件的恶意代码分析:使用SystemInternal工具与内核调试器研究windows用户空间与内核空间 ******************** 既然本篇的主角是PE文件,那 ...

mahout基于hadoop的CF代码分析

mahout基于hadoop的CF代码分析相关推荐

最新文章

热门文章