Mahout 协同过滤 itemBase RecommenderJob源码分析

来自：http://blog.csdn.net/heyutao007/article/details/8612906

Mahout支持2种 M/R 的jobs实现itemBase的协同过滤

I.ItemSimilarityJob

II.RecommenderJob

下面我们对RecommenderJob进行分析，版本是mahout-distribution-0.7

源码包位置：org.apache.mahout.cf.taste.hadoop.item.RecommenderJob

RecommenderJob前几个阶段和ItemSimilarityJob是一样的，不过ItemSimilarityJob
计算出item的相似度矩阵就结束了，而RecommenderJob 会继续使用相似度矩阵，对每个user计算出应该推荐给他的top N
个items。RecommenderJob 的输入也是userID, itemID[,
preferencevalue]格式的。JobRecommenderJob主要由以下一系列的Job组成：

1 PreparePreferenceMatrixJob（同ItemSimilarityJob）

输入： (userId, itemId, pref)

1.1 itemIDIndex 将Long型的itemID转成一个int型的index

1.2 toUserVectors 将输入的 (userId, itemId, pref) 转成user向量 USER_VECTORS (userId, VectorWritable<itemId, pref>)

1.3 toItemVectors 使用 USER_VECTORS 构建item向量 RATING_MATRIX (itemId,VectorWritable<userId,pref>)

2 RowSimilarityJob（同ItemSimilarityJob）

2.1 normsAndTranspose

计算每个item的norm，并转成user向量

输入：RATING_MATRIX

（1）使用similarity.normalize处理每个item向量，使用similarity.norm计算每个item的norm，写到hdfs；

（2）根据item向量进行转置，即输入：item-（user，pref），输出：user-（item，pref）。这一步的目的是将同一个user
喜欢的item对找出来，因为只有两个item有相同的user喜欢，我们才认为它们是相交的，下面才有对它们计算相似度的必要。

2.2 pairwiseSimilarity

计算item对之间的相似度

输入：2.1（2）计算出的user向量user-（item，pref）

map：CooccurrencesMapper

使用一个两层循环，对user向量中两两item，以itemM为key，所有itemM之后的itemN与 itemM的similarity.aggregate计算值组成的向量为value。

reduce：SimilarityReducer

（1）叠加相同的两个item在不同用户之间的aggregate值，得到itemM-（（ item M+1， aggregate M+1），（
item M+2， aggregate M+2），（ item M+3， aggregate M+3）。。。）

（2）然后计算itemM和之后所有item之间的相似度。相似度计算使用similarity.similarity，第一个参数是两个item的
aggregate值，后两个参数是两个item的norm值，norm值在上一个Job已经得到。结果是以itemM为key，所有itemM之后的
itemN与 itemM 相似度组成的向量为value，即itemM-（（ item M+1， simi M+1），（ item M+2，
simi M+2），（ item M+3， simi M+ 3）。。。）

到这里我们实际上是得到了相似度矩阵的斜半部分。

2.3 asMatrix

构造完整的相似度矩阵（上面得到的只是一个斜半部分）

输入：2.2reduce（2）输出的以itemM为key，所有itemM之后的itemN与之相似度组成的向量

map：UnsymmetrifyMapper

（1）反转，根据item M-（item M+1，simiM+1）记录item M+1 -（item M，simiM+1）

（2）使用一个优先队列求出itemM的top maxSimilaritiesPerRow（可设置参数）个相似item，比如maxSimilaritiesPerRow =2时，可能输出

itemM-（（ item M+1， simi M+1），（ item M+3， simi M+3））

reduce：MergeToTopKSimilaritiesReducer

（1）对相同的item M，合并上面两种向量，这样就形成了完整的相似度矩阵，itemM-（（ item 1， simi 1），（ item 2， simi 2））。。。，（ item N， simi N））。

（2）使用Vectors.topKElements对每个item求top maxSimilaritiesPerRow（可设置参数）个相似item。可见map（2）中的求topN是对这一步的一个预先优化。

最终输出的是itemM-（（ item A， simi A），（ item B， simi B））。。。，（ item N， simi N））,A到N的个数是maxSimilaritiesPerRow。

至此RowSimilarityJob结束。下面就进入了和ItemSimilarityJob不同的地方。

3 prePartialMultiply1 + prePartialMultiply2 + partialMultiply

这三个job的工作是将1.2生成的user向量和2.3reduce（2）生成的相似度矩阵使用相同的item作为key聚合到一起，实际上是为下面会
提到的矩阵乘法做准备。VectorOrPrefWritable是两种value的统一结构，它包含了相似度矩阵中某个item的一列和user向量中
对应这个item的（userID， prefValue ）。

[java] view plaincopy

public final class VectorOrPrefWritable implements Writable {

private Vector vector;

private long userID;

private float value;

}

下面依次介绍：

3.1 prePartialMultiply1

输入：2.3reduce（2）生成的相似度矩阵。

以item为key，相似度矩阵的一行包装成一个VectorOrPrefWritable为value。矩阵相乘应该使用列，但是对于相似度矩阵，行和列是一样的。

3.2 prePartialMultiply2

输入：1.2生成的USER_VECTORS

对user，以每个item为key，userID和对应这个item的prefValue包装成一个VectorOrPrefWritable为value。

3.3 partialMultiply

以3.1和3.2的输出为输入，聚合到一起，生成item为key，VectorAndPrefsWritable为value为value。
VectorAndPrefsWritable包含了相似度矩阵中某个item一列和一个List<Long>
userIDs，一个List<Float> values。

[java] view plaincopy

public final class VectorAndPrefsWritable implements Writable {

private Vector vector;

private List<Long> userIDs;

private List<Float> values;

}

4 itemFiltering

用户设置过滤某些user，需要将user/item pairs也转成（itemID，VectorAndPrefsWritable）形式

5 aggregateAndRecommend

一切就绪后，下面就开始计算推荐向量了。推荐计算公式如下：

Prediction(u,i) = sum(all n from N: similarity(i,n) * rating(u,n)) / sum(all n from N: abs(similarity(i,n)))

u = a user

i = an item not yet rated by u

N = all items similar to i

可以看到，分子部分就是一个相似度矩阵和user向量的矩阵乘法。对于这个矩阵乘法，实现代码和传统的矩阵乘法不一样，其伪代码：

assign R to be the zero vector

for each column i in the co-occurrence matrix

multiply column vector i by the ith element of the user vector

add this vector to R

假设相似度矩阵的大小是N，则以上代码实际上是对某个user的所有item，将这个item在相似度矩阵中对应列和user对这个item的
prefValue相乘，得到N个向量后，再将这些向量相加，就得到了针对这个用户的N个item的推荐向量。要实现这些，首先要把某个user对所有
item的prefValue以及这个item在相似度矩阵中对应列聚合到一起。下面看实现：

输入：3.3和4的输出

map：PartialMultiplyMapper

将（itemID，VectorAndPrefsWritable）形式转成以userID为
key，PrefAndSimilarityColumnWritable为value。
PrefAndSimilarityColumnWritable包含了这个user对一个item的prefValue和item在相似度矩阵中的那
列，其实还是使用的VectorAndPrefsWritable中的vector和value。

[java] view plaincopy

public final class PrefAndSimilarityColumnWritable implements Writable {

private float prefValue;

private Vector similarityColumn;

}

reduce：AggregateAndRecommendReducer

收集到属于这个user的所有 PrefAndSimilarityColumnWritable 后，下面就是进行矩阵相乘的工作。

根据是否设置booleanData，有以下两种操作：

（1）reduceBooleanData

只是单纯的将所有的PrefAndSimilarityColumnWritable 中的SimilarityColumn相加，没有用到item-pref。

（2）reduceNonBooleanData

用到item-pref的计算方法，

分子部分，是矩阵相乘的结果，根据上面的伪代码，它是将每个PrefAndSimilarityColumnWritable
中的SimilarityColumn和 prefValue
的相乘，生成多个向量后再将这些向量相加；而分母是所有的SimilarityColumn和。下面看代码：

代码：

[java] view plaincopy

for (PrefAndSimilarityColumnWritable prefAndSimilarityColumn : values) {

Vector simColumn = prefAndSimilarityColumn.getSimilarityColumn();

float prefValue = prefAndSimilarityColumn.getPrefValue();

//分子部分，每个SimilarityColumn和item-pref的乘积生成多个向量，然后将这些向量相加

numerators = numerators == null

? prefValue == BOOLEAN_PREF_VALUE ? simColumn.clone() : simColumn.times(prefValue)

: numerators.plus(prefValue == BOOLEAN_PREF_VALUE ? simColumn : simColumn.times(prefValue));

simColumn.assign(ABSOLUTE_VALUES);

//分母是所有的SimilarityColumn和

denominators = denominators == null ? simColumn : denominators.plus(simColumn);

}

两者相除，就得到了反映推荐可能性的数值。

之后还会使用writeRecommendedItems使用一个优先队列取top推荐，并且将index转成真正的itemID，最终完成。

在以上分析中，similarity是一个VectorSimilarityMeasure接口实现，它是一个相似度算法接口，主要方法有：

（1）Vector normalize(Vector vector);

（2）double norm(Vector vector);

（3）double aggregate(double nonZeroValueA, double nonZeroValueB);

（4）double similarity(double summedAggregations, double normA, double normB, int numberOfColumns);

（5）boolean consider(int numNonZeroEntriesA, int numNonZeroEntriesB, double maxValueA, double maxValueB,

double threshold);

众多的相似度算法就是实现了这个接口，比如TanimotoCoefficientSimilarity的similarity实现就是：

public double similarity(double dots, double normA, double normB, int numberOfColumns) {

return dots / (normA + normB - dots);

}