mahout算法源码分析之Collaborative Filtering with ALS-WR （三）parallelALS initializeM和for循环

runSolver(pathToUserRatings(), pathToU(currentIteration), pathToM(currentIteration - 1));

private void runSolver(Path ratings, Path output, Path pathToUorI)
      throws ClassNotFoundException, IOException, InterruptedException {

    Class<? extends Mapper> solverMapper = implicitFeedback ?
        SolveImplicitFeedbackMapper.class : SolveExplicitFeedbackMapper.class;

    Job solverForUorI = prepareJob(ratings, output, SequenceFileInputFormat.class, solverMapper, IntWritable.class,
        VectorWritable.class, SequenceFileOutputFormat.class);
    Configuration solverConf = solverForUorI.getConfiguration();
    solverConf.set(LAMBDA, String.valueOf(lambda));
    solverConf.set(ALPHA, String.valueOf(alpha));
    solverConf.setInt(NUM_FEATURES, numFeatures);
    solverConf.set(FEATURE_MATRIX, pathToUorI.toString());
    boolean succeeded = solverForUorI.waitForCompletion(true);
    if (!succeeded) 
      throw new IllegalStateException("Job failed!");
  }

package mahout.fansy.als;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
import org.apache.mahout.math.SequentialAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.mahout.math.als.AlternatingLeastSquaresSolver;
import org.apache.mahout.math.map.OpenIntObjectHashMap;

import com.google.common.collect.Lists;

import mahout.fansy.utils.read.ReadArbiKV;

public class SolveExplicitFeedbackMapperFollow {

	/**
	 * 第一次调用SloveExplicitFeedBackMapper的仿制代码
	 * @param args 
	 */
	private static double lambda=0.065;
    private static int numFeatures=20;

    private static OpenIntObjectHashMap<Vector> UorM;

    private static AlternatingLeastSquaresSolver solver;
	public static void main(String[] args) throws IOException {
		setup();
		map();
	}
	
	/**
	 * 获得map输入文件;
	 * @return
	 * @throws IOException
	 */
	public static Map<Writable,Writable> getMapData() throws IOException{
		String fPath="hdfs://ubuntu:9000/user/mahout/output/als/als/userRatings/part-r-00000";
		Map<Writable,Writable> mapData=ReadArbiKV.readFromFile(fPath);
		return mapData;
	}
	/**
	 * 仿造setup函数
	 */
	public static void setup(){
		solver = new AlternatingLeastSquaresSolver();
		UorM = ALSUtilsFollow.readMatrixByRows(
				new Path("hdfs://ubuntu:9000/user/mahout/temp/als/M--1/part-m-00000"), getConf());
	}

	public static void map() throws IOException{
		Map<Writable,Writable> map=getMapData();
		for(Iterator<Entry<Writable, Writable>> iter=map.entrySet().iterator();iter.hasNext();){
			Entry<Writable,Writable> entry=(Entry<Writable, Writable>) iter.next();
			IntWritable userOrItemID=(IntWritable) entry.getKey();
			VectorWritable ratingsWritable=(VectorWritable) entry.getValue();
			// source code
			Vector ratings = new SequentialAccessSparseVector(ratingsWritable.get());
		      List<Vector> featureVectors = Lists.newArrayList();
		      Iterator<Vector.Element> interactions = ratings.iterateNonZero();
		      while (interactions.hasNext()) {
		        int index = interactions.next().index();
		        featureVectors.add(UorM.get(index));
		      }

		      Vector uiOrmj = solver.solve(featureVectors, ratings, lambda, numFeatures);

		      System.out.println(userOrItemID+","+ new VectorWritable(uiOrmj));
		}
	}
	/**
	 * 获得configuration
	 * @return
	 */
	private static Configuration getConf() {
		Configuration conf=new Configuration();
		conf.set("mapred.job.tracker", "ubuntu:9000");
		return conf;
	}
}

这里方框中可以看到有3692个item，和前面分析一致，同时每个都含有20个值，从0到19，这个把值往后面拉就可以看到了。

首先由userID所有评价过的item找出在UorM中存在的项，然后把这些项全部赋值给featureVector，即可得到userID的featureVector了。

public Vector solve(Iterable<Vector> featureVectors, Vector ratingVector, double lambda, int numFeatures) {

    Preconditions.checkNotNull(featureVectors, "Feature vectors cannot be null");
    Preconditions.checkArgument(!Iterables.isEmpty(featureVectors));
    Preconditions.checkNotNull(ratingVector, "rating vector cannot be null");
    Preconditions.checkArgument(ratingVector.getNumNondefaultElements() > 0, "Rating vector cannot be empty");
    Preconditions.checkArgument(Iterables.size(featureVectors) == ratingVector.getNumNondefaultElements());

    int nui = ratingVector.getNumNondefaultElements();

    Matrix MiIi = createMiIi(featureVectors, numFeatures);
    Matrix RiIiMaybeTransposed = createRiIiMaybeTransposed(ratingVector);

    /* compute Ai = MiIi * t(MiIi) + lambda * nui * E */
    Matrix Ai = addLambdaTimesNuiTimesE(MiIi.times(MiIi.transpose()), lambda, nui);
    /* compute Vi = MiIi * t(R(i,Ii)) */
    Matrix Vi = MiIi.times(RiIiMaybeTransposed);
    /* compute Ai * ui = Vi */
    return solve(Ai, Vi);
  }

其中第一行全部是相应item的平均评价值，下面的行则是随机数评价值。然后到了Matrix = createRiIiMaybeTransposed(ratingVector);这里的

接下来到了Matrix Ai = addLambdaTimesNuiTimesE(MiIi.times(MiIi.transpose()), lambda, nui);即Ai的值其实是等于MiIi的转置乘以MiIi，这里的乘法是矩阵的乘法。比如行矩阵[a1,a2,a3]乘以列矩阵[b1;b2;b3]，那么即是1*3的行矩阵乘以3*1的列矩阵，得到1*1的矩阵：a1*b1+a2*b2+a3*b3。通俗来说应该是M*R的矩阵乘以R*N的矩阵，然后得到M*N的矩阵。Mili.transpose()函数就是求Mili的转置，times是求两个矩阵相乘。

Vector solve(Matrix Ai, Matrix Vi) {
    return new QRDecomposition(Ai).solve(Vi).viewColumn(0);
  }

http://blog.csdn.net/fansy1990