ParallelALSFactorizationJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption("lambda", "l", "", true); addOption("numFeatures", "f", "", true); addOption("numIterations", "i", "", true); Map<String,String> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } int numFeatures = Integer.parseInt(parsedArgs.get("--numFeatures")); int numIterations = Integer.parseInt(parsedArgs.get("--numIterations")); double lambda = Double.parseDouble(parsedArgs.get("--lambda")); tempDir = parsedArgs.get("--tempDir"); Job itemRatings = prepareJob(getInputPath(), pathToItemRatings(), TextInputFormat.class, PrefsToRatingsMapper.class, VarIntWritable.class, FeatureVectorWithRatingWritable.class, Reducer.class, VarIntWritable.class, FeatureVectorWithRatingWritable.class, SequenceFileOutputFormat.class); itemRatings.waitForCompletion(true); Job userRatings = prepareJob(getInputPath(), pathToUserRatings(), TextInputFormat.class, PrefsToRatingsMapper.class, VarIntWritable.class, FeatureVectorWithRatingWritable.class, Reducer.class, VarIntWritable.class, FeatureVectorWithRatingWritable.class, SequenceFileOutputFormat.class); userRatings.getConfiguration().setBoolean(MAP_TRANSPOSED, Boolean.TRUE); userRatings.waitForCompletion(true); Job initializeM = prepareJob(getInputPath(), pathToM(-1), TextInputFormat.class, ItemIDRatingMapper.class, VarLongWritable.class, FloatWritable.class, InitializeMReducer.class, VarIntWritable.class, FeatureVectorWithRatingWritable.class, SequenceFileOutputFormat.class); initializeM.getConfiguration().setInt(NUM_FEATURES, numFeatures); initializeM.waitForCompletion(true); for (int n = 0; n < numIterations; n++) { iterate(n, numFeatures, lambda); } Job uAsMatrix = prepareJob(pathToU(numIterations - 1), new Path(getOutputPath(), "U"), SequenceFileInputFormat.class, ToMatrixMapper.class, IntWritable.class, VectorWritable.class, Reducer.class, IntWritable.class, VectorWritable.class, SequenceFileOutputFormat.class); uAsMatrix.waitForCompletion(true); Job mAsMatrix = prepareJob(pathToM(numIterations - 1), new Path(getOutputPath(), "M"), SequenceFileInputFormat.class, ToMatrixMapper.class, IntWritable.class, VectorWritable.class, Reducer.class, IntWritable.class, VectorWritable.class, SequenceFileOutputFormat.class); mAsMatrix.waitForCompletion(true); return 0; }  static void main(String[] args); @Override int run(String[] args);  }
@Test public void completeJobToyExample() throws Exception { File inputFile = getTestTempFile("prefs.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Double na = Double.NaN; Matrix preferences = new SparseRowMatrix(new int[] { 4, 4 }, new Vector[] { new DenseVector(new double[] {5.0, 5.0, 2.0, na }), new DenseVector(new double[] {2.0, na, 3.0, 5.0 }), new DenseVector(new double[] { na, 5.0, na, 3.0 }), new DenseVector(new double[] {3.0, na, na, 5.0 }) }); StringBuilder prefsAsText = new StringBuilder(); String separator = ""; Iterator<MatrixSlice> sliceIterator = preferences.iterateAll(); while (sliceIterator.hasNext()) { MatrixSlice slice = sliceIterator.next(); Iterator<Vector.Element> elementIterator = slice.vector().iterateNonZero(); while (elementIterator.hasNext()) { Vector.Element e = elementIterator.next(); if (!Double.isNaN(e.get())) { prefsAsText.append(separator).append(slice.index()).append(',').append(e.index()).append(',').append(e.get()); separator = "\n"; } } } logger.info("Input matrix:\n" + prefsAsText); writeLines(inputFile, prefsAsText.toString()); ParallelALSFactorizationJob alsFactorization = new ParallelALSFactorizationJob(); Configuration conf = new Configuration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); alsFactorization.setConf(conf); int numFeatures = 3; int numIterations = 5; double lambda = 0.065; alsFactorization.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--lambda", String.valueOf(lambda), "--numFeatures", String.valueOf(numFeatures), "--numIterations", String.valueOf(numIterations) }); Matrix u = MathHelper.readEntries(conf, new Path(outputDir.getAbsolutePath(), "U/part-r-00000"), preferences.numRows(), numFeatures); Matrix m = MathHelper.readEntries(conf, new Path(outputDir.getAbsolutePath(), "M/part-r-00000"), preferences.numCols(), numFeatures); RunningAverage avg = new FullRunningAverage(); sliceIterator = preferences.iterateAll(); while (sliceIterator.hasNext()) { MatrixSlice slice = sliceIterator.next(); Iterator<Vector.Element> elementIterator = slice.vector().iterateNonZero(); while (elementIterator.hasNext()) { Vector.Element e = elementIterator.next(); if (!Double.isNaN(e.get())) { double pref = e.get(); double estimate = u.getRow(slice.index()).dot(m.getRow(e.index())); double err = pref - estimate; avg.addDatum(err * err); logger.info("Comparing preference of user [" + slice.index() + "] towards item [" + e.index() + "], " + "was [" + pref + "] estimate is [" + estimate + ']'); } } } double rmse = Math.sqrt(avg.getAverage()); logger.info("RMSE: " + rmse); assertTrue(rmse < 0.2); }