mahout算法源码分析之Itembased Collaborative Filtering（四）共生矩阵乘法

Mahout版本：0.7，hadoop版本：1.0.4，jdk：1.7.0_25 64bit。

经过了SimilarityJob的计算共生矩阵后，就可以开始下面一个过程了，这个过程主要是共生矩阵的乘法，要说这个共生矩阵的乘法是啥意思？我也不是很清楚，不清楚就看代码呗。

首先明确共生矩阵，即共生矩阵的输入文件（也是上面个SimilarityJob的输出文件）：

similarityMatrix=================  
{102={101:0.14201473202245876,106:0.14972506706560876,105:0.14328432723886902,104:0.12789210656028413,103:0.1975496259559987}, 
103={101:0.15548737703860027,106:0.1424339656566283,105:0.11208890297777215,104:0.14037600977966974,102:0.1975496259559987},
 101={107:0.10275248635596666,106:0.1424339656566283,105:0.1158457425543559,104:0.16015261286229274,103:0.15548737703860027,102:0.14201473202245876},
106={101:0.1424339656566283,105:0.14201473202245876,104:0.18181818181818182,103:0.1424339656566283,102:0.14972506706560876}, 
107={105:0.2204812092115424,104:0.13472338607037426,101:0.10275248635596666}, 
104={107:0.13472338607037426,106:0.18181818181818182,105:0.16736577623297264,103:0.14037600977966974,102:0.12789210656028413,101:0.16015261286229274}, 
105={107:0.2204812092115424,106:0.14201473202245876,104:0.16736577623297264,103:0.11208890297777215,102:0.14328432723886902,101:0.1158457425543559}}  

计算共生矩阵一共分为三个job，下面来分别看：

（1）prePartialMultiply1：
这个job的调用代码如下：

Job prePartialMultiply1 = prepareJob(
              similarityMatrixPath, prePartialMultiplyPath1, SequenceFileInputFormat.class,
              SimilarityMatrixRowWrapperMapper.class, VarIntWritable.class, VectorOrPrefWritable.class,
              Reducer.class, VarIntWritable.class, VectorOrPrefWritable.class,
              SequenceFileOutputFormat.class);

可以看到其实只是有一个mapper而已，reducer形同虚设，那就只看mapper吧

（1.1）mapper：//SimilarityMatrixRowWrapperMapper

这个mapper同样很简单，只有一个map函数，如下：

（1.1.2）map：

protected void map(IntWritable key,
                     VectorWritable value,
                     Context context) throws IOException, InterruptedException {
    Vector similarityMatrixRow = value.get();
    /* remove self similarity */
    similarityMatrixRow.set(key.get(), Double.NaN);
    context.write(new VarIntWritable(key.get()), new VectorOrPrefWritable(similarityMatrixRow));
  }

map中首先获得输入的value，就是和项目key不同的其他项目的相似度向量；然后是把自身的相似度设置为最大，在前面similarityJob中最后是把自身的相似度设置为0的，这里又设置为最大（前面设置为0，应该会在输出文件中就会相应的占少点空间，然后这里设置为最大值是为了后面计算的需要吧），然后就把key和value输出了，这里看到key和value都进行了类型转换，其中的key的类型变为了VarIntWritable应该只是为了速度的考虑吧（为啥之前不用？），而value的类型转为VectorOrPrefWritable应该是为了后面转换的考虑；

（1.2）reducer：//Reducer

protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context
                        ) throws IOException, InterruptedException {
    for(VALUEIN value: values) {
      context.write((KEYOUT) key, (VALUEOUT) value);
    }
  }

等于就是把mapper中的重新输出一遍而已；

所以他的输出应该类似于：

{102={106:0.1497250646352768,105:0.14328432083129883,104:0.12789210677146912,103:0.19754962623119354,102:NaN,101:0.14201472699642181},
 103={106:0.14243397116661072,105:0.11208890378475189,104:0.140376016497612,103:NaN,102:0.19754962623119354,101:0.15548737347126007}, 
101={107:0.10275248438119888,106:0.14243397116661072,105:0.11584573984146118,104:0.1601526141166687,103:0.15548737347126007,102:0.14201472699642181,101:NaN},
 106={106:NaN,105:0.14201472699642181,104:0.1818181872367859,103:0.14243397116661072,102:0.1497250646352768,101:0.14243397116661072}, 
107={101:0.10275248438119888,107:NaN,105:0.22048120200634003,104:0.13472338020801544}, 

104={107:0.13472338020801544,106:0.1818181872367859,105:0.16736577451229095,104:NaN,103:0.140376016497612,102:0.12789210677146912,101:0.1601526141166687}, 105={107:0.22048120200634003,106:0.14201472699642181,105:NaN,104:0.16736577451229095,103:0.11208890378475189,102:0.14328432083129883,101:0.11584573984146118}}

（2）prePartialMultiply2：

首先是其调用代码：

Job prePartialMultiply2 = prepareJob(new Path(prepPath, PreparePreferenceMatrixJob.USER_VECTORS),
              prePartialMultiplyPath2, SequenceFileInputFormat.class, UserVectorSplitterMapper.class, VarIntWritable.class,
              VectorOrPrefWritable.class, Reducer.class, VarIntWritable.class, VectorOrPrefWritable.class,
              SequenceFileOutputFormat.class);
      if (usersFile != null) {
        prePartialMultiply2.getConfiguration().set(UserVectorSplitterMapper.USERS_FILE, usersFile);
      }
      prePartialMultiply2.getConfiguration().setInt(UserVectorSplitterMapper.MAX_PREFS_PER_USER_CONSIDERED,
              maxPrefsPerUser);

这里看到同样的也只是有mapper在起作用而已，reducer同样的形同虚设；然后就是可以提供一个usersFile文件用于过滤不关心的用户，这个在这里没有设置，默认是null的，同样后面的maxPrefsPerUser也没有设置，所以也是默认值；

首先来明确一下这里的输入文件，这里的输入是第一个PreparePreferenceMatrixJob中的userVecotr的输出，根据前面的分析，得到文件如下：

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

（2.1）mapper：//UserVectorSplitterMapper

这个mapper是有setup和map函数的，下面一个个分析：

（2.1.1）setup：

protected void setup(Context context) throws IOException {
    Configuration jobConf = context.getConfiguration();
    maxPrefsPerUserConsidered = jobConf.getInt(MAX_PREFS_PER_USER_CONSIDERED, DEFAULT_MAX_PREFS_PER_USER_CONSIDERED);
    String usersFilePathString = jobConf.get(USERS_FILE);
    if (usersFilePathString != null) {
      FSDataInputStream in = null;
      try {
        Path unqualifiedUsersFilePath = new Path(usersFilePathString);
        FileSystem fs = FileSystem.get(unqualifiedUsersFilePath.toUri(), jobConf);
        usersToRecommendFor = new FastIDSet();
        Path usersFilePath = unqualifiedUsersFilePath.makeQualified(fs);
        in = fs.open(usersFilePath);
        for (String line : new FileLineIterable(in)) {
          try {
            usersToRecommendFor.add(Long.parseLong(line));
          } catch (NumberFormatException nfe) {
            log.warn("usersFile line ignored: {}", line);
          }
        }
      } finally {
        Closeables.closeQuietly(in);
      }
    }
  }

汗，写了这么多代码，如果不用过滤用户的话，那么if里面的都没用了。不过，有时有一些特殊的需要（不关系其他用户，提高效率），这样写还是很不错的。其实看if里面做的事情也只是把文件中的user读出来，然后放在一个类似集合里面，然后在map中做过滤而已。除了if里面的代码，在setup中基本就没做其他事情了，因为那个maxPrefsPerUser没有设置，按默认的来，所以。。。

（2.1.2）map：

protected void map(VarLongWritable key,
                     VectorWritable value,
                     Context context) throws IOException, InterruptedException {
    long userID = key.get();
    if (usersToRecommendFor != null && !usersToRecommendFor.contains(userID)) {
      return;
    }
    Vector userVector = maybePruneUserVector(value.get());
    Iterator<Vector.Element> it = userVector.iterateNonZero();
    VarIntWritable itemIndexWritable = new VarIntWritable();
    VectorOrPrefWritable vectorOrPref = new VectorOrPrefWritable();
    while (it.hasNext()) {
      Vector.Element e = it.next();
      itemIndexWritable.set(e.index());
      vectorOrPref.set(userID, (float) e.get());
      context.write(itemIndexWritable, vectorOrPref);
    }
  }

map的内容也不是很多，首先把userid读出来，然后在userToRecommendFor（就是在setup中读文件读到的user，这里没有而已）中过滤这个userid，如果没有就直接返回。接下来是对输入的value进行处理了，看下这个maybePruneUserVector函数是干嘛的：

private Vector maybePruneUserVector(Vector userVector) {
    if (userVector.getNumNondefaultElements() <= maxPrefsPerUserConsidered) {
      return userVector;
    }

    float smallestLargeValue = findSmallestLargeValue(userVector);

    // "Blank out" small-sized prefs to reduce the amount of partial products
    // generated later. They're not zeroed, but NaN-ed, so they come through
    // and can be used to exclude these items from prefs.
    Iterator<Vector.Element> it = userVector.iterateNonZero();
    while (it.hasNext()) {
      Vector.Element e = it.next();
      float absValue = Math.abs((float) e.get());
      if (absValue < smallestLargeValue) {
        e.set(Float.NaN);
      }
    }
    return userVector;
  }

先看if条件判断，因为传入的vector最大的size也只是7（一共有7个项目），所以这里判断是true的，那么就直接返回了原来的vector了。但是现实中一般的项目数肯定是大于10的，所以，如果maxPrefsPerUser是按默认值的话，就会执行下面的代码，而不是直接返回原始向量，那么分析下吧。紧接着是一个findSmallestLargeValue函数：

private float findSmallestLargeValue(Vector userVector) {

    TopK<Float> topPrefValues = new TopK<Float>(maxPrefsPerUserConsidered, new Comparator<Float>() {
      @Override
      public int compare(Float one, Float two) {
        return one.compareTo(two);
      }
    });

    Iterator<Vector.Element> it = userVector.iterateNonZero();
    while (it.hasNext()) {
      float absValue = Math.abs((float) it.next().get());
      topPrefValues.offer(absValue);
    }
    return topPrefValues.smallestGreat();
  }

目测这个函数应该是输出最小的float值，做个测试吧：

package mahout.fansy.item;

import java.util.Comparator;
import java.util.Iterator;

import org.apache.mahout.cf.taste.common.TopK;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;

import junit.framework.TestCase;
/*
 * 测试findSmallestLargeValue方法；
 */
public class TestTopK extends TestCase {
	private Vector v=null;
	public void setup(){
		v=new RandomAccessSparseVector(Integer.MAX_VALUE);
		//{103:2.5,102:3.0,101:5.0}
		v.set(101, 5.0);
		v.set(102,3.0);
		v.set(103,2.5);
	}
	
	public void testTopK(){
		this.setup();
		 TopK<Float> topPrefValues = new TopK<Float>(10, new Comparator<Float>() {
		      @Override
		      public int compare(Float one, Float two) {
		        return one.compareTo(two);
		      }
		    });

		    Iterator<Vector.Element> it = v.iterateNonZero();
		    while (it.hasNext()) {
		      float absValue = Math.abs((float) it.next().get());
		      topPrefValues.offer(absValue);
		    }
		    System.out.println("the value is:"+topPrefValues.smallestGreat());
	}
}

由上面的输出结果2.5来看，的确是最小的值了；maybePruneUserVector中后面的代码就不是很理解了，既然都找出了最小值了，那么怎么还会有最小值呢？看他的英文解释好像是说要过滤一些什么输出似的，但是好像这样过滤不了吧？疑问？？？

如果真的是按照我想的那样的话，那么uservector还是输出原始值的；接下来的map中的代码和以前的分析中有些内容很相似，针对这样的输入<key,value>  -->  <userid,vector[itemid1:prefValue1,itemid2:prefValue2]> 输出为 <itemid1,vector[userid:prefValue1]>   、<itemid2,vecotr[userid:prefVlaue2]>，同样的，格式也是经过设置的。

(2.2)reducer://Reducer也就是把mapper的输出结果重复输出一遍，没有其他操作。

（3）partialMultiply：

其调用代码如下：

Job partialMultiply = prepareJob(
              new Path(prePartialMultiplyPath1 + "," + prePartialMultiplyPath2), partialMultiplyPath,
              SequenceFileInputFormat.class, Mapper.class, VarIntWritable.class, VectorOrPrefWritable.class,
              ToVectorAndPrefReducer.class, VarIntWritable.class, VectorAndPrefsWritable.class,
              SequenceFileOutputFormat.class);
      setS3SafeCombinedInputPath(partialMultiply, getTempPath(), prePartialMultiplyPath1, prePartialMultiplyPath2);

先看下setS3SafeCombinedInputPath的操作是什么，

 public static void setS3SafeCombinedInputPath(Job job, Path referencePath, Path inputPathOne, Path inputPathTwo)
      throws IOException {
    FileSystem fs = FileSystem.get(referencePath.toUri(), job.getConfiguration());
    FileInputFormat.setInputPaths(job, inputPathOne.makeQualified(fs), inputPathTwo.makeQualified(fs));
  }

根据英文解释是说这样可以支持在Amazon S3上运行，因为MultipleInputs还没有投入使用。姑且认为这个是设置两个输入路径合法吧。

这个job有mapper也有reducer，额，好吧，这次该mapper形同虚设了，下面一个个分析：

先说下输入输入格式吧，输入数据有两种状态，（数据的类型是一样的）：

其一，  <itemid, user:prefValue>； 其二，< itemid ，[itemid:prefValue,itemid:prefValue]>。

（3.1）mapper：//Mapper，不解释。。。

（3.2）reducer：//ToVectorAndPrefReducer

（3.2.1）reduce：

protected void reduce(VarIntWritable key,
                        Iterable<VectorOrPrefWritable> values,
                        Context context) throws IOException, InterruptedException {

    List<Long> userIDs = Lists.newArrayList();
    List<Float> prefValues = Lists.newArrayList();
    Vector similarityMatrixColumn = null;
    for (VectorOrPrefWritable value : values) {
      if (value.getVector() == null) {
        // Then this is a user-pref value
        userIDs.add(value.getUserID());
        prefValues.add(value.getValue());
      } else {
        // Then this is the column vector
        if (similarityMatrixColumn != null) {
          throw new IllegalStateException("Found two similarity-matrix columns for item index " + key.get());
        }
        similarityMatrixColumn = value.getVector();
      }
    }

    if (similarityMatrixColumn == null) {
      return;
    }

    VectorAndPrefsWritable vectorAndPrefs = new VectorAndPrefsWritable(similarityMatrixColumn, userIDs, prefValues);
    context.write(key, vectorAndPrefs);
  }

reduce操作其实就等于是把相同的key的所有属性放到一个变量中，比如item是101的vector为 101={107:0.10275248635596666,106:0.1424339656566283,105:0.1158457425543559,104:0.16015261286229274,103:0.15548737703860027,102:0.14201473202245876}，即similarityMatrixColumn变量；然后userIDs就是101={5:4.0,4:5.0,3:2.5,2:2.0,1:5.0}其中的userids={5,4,3,2,1}，prefValues={4.0,5.0,2.5,2.0,5.0}.

测试输入的结果如下：

{102={106:0.1497250646352768,105:0.14328432083129883,104:0.12789210677146912,103:0.19754962623119354,102:NaN,101:0.14201472699642181}	[5, 1, 2]	[3.0, 3.0, 2.5], 
103={106:0.14243397116661072,105:0.11208890378475189,104:0.140376016497612,103:NaN,102:0.19754962623119354,101:0.15548737347126007}	[4, 1, 2, 5]	[3.0, 2.5, 5.0, 2.0], 
101={107:0.10275248438119888,106:0.14243397116661072,105:0.11584573984146118,104:0.1601526141166687,103:0.15548737347126007,102:0.14201472699642181,101:NaN}	[5, 1, 4, 2, 3]	[4.0, 5.0, 5.0, 2.0, 2.5], 
106={106:NaN,105:0.14201472699642181,104:0.1818181872367859,103:0.14243397116661072,102:0.1497250646352768,101:0.14243397116661072}	[4, 5]	[4.0, 4.0], 

107={101:0.10275248438119888,107:NaN,105:0.22048120200634003,104:0.13472338020801544}	[3]	[5.0], 
104={107:0.13472338020801544,106:0.1818181872367859,105:0.16736577451229095,104:NaN,103:0.140376016497612,102:0.12789210677146912,101:0.1601526141166687}	[4, 2, 5, 3]	[4.5, 2.0, 4.0, 4.0], 
105={107:0.22048120200634003,106:0.14201472699642181,105:NaN,104:0.16736577451229095,103:0.11208890378475189,102:0.14328432083129883,101:0.11584573984146118}	[5, 3]	[3.5, 4.5]}

从上面的文件读取结果来看，分析思路是正确的。

分享，成长，快乐

转载请注明blog地址：http://blog.csdn.net/fansy1990