ParallelALSFactorizationJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption("lambda", null, "regularization parameter", true); addOption("implicitFeedback", null, "data consists of implicit feedback?", String.valueOf(false)); addOption("alpha", null, "confidence parameter (only used on implicit feedback)", String.valueOf(40)); addOption("numFeatures", null, "dimension of the feature space", true); addOption("numIterations", null, "number of iterations", true); addOption("numThreadsPerSolver", null, "threads per solver mapper", String.valueOf(1)); addOption("usesLongIDs", null, "input contains long IDs that need to be translated"); Map<String,List<String>> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } numFeatures = Integer.parseInt(getOption("numFeatures")); numIterations = Integer.parseInt(getOption("numIterations")); lambda = Double.parseDouble(getOption("lambda")); alpha = Double.parseDouble(getOption("alpha")); implicitFeedback = Boolean.parseBoolean(getOption("implicitFeedback")); numThreadsPerSolver = Integer.parseInt(getOption("numThreadsPerSolver")); boolean usesLongIDs = Boolean.parseBoolean(getOption("usesLongIDs", String.valueOf(false))); if (usesLongIDs) { Job mapUsers = prepareJob(getInputPath(), getOutputPath("userIDIndex"), TextInputFormat.class, MapLongIDsMapper.class, VarIntWritable.class, VarLongWritable.class, IDMapReducer.class, VarIntWritable.class, VarLongWritable.class, SequenceFileOutputFormat.class); mapUsers.getConfiguration().set(TOKEN_POS, String.valueOf(TasteHadoopUtils.USER_ID_POS)); mapUsers.waitForCompletion(true); Job mapItems = prepareJob(getInputPath(), getOutputPath("itemIDIndex"), TextInputFormat.class, MapLongIDsMapper.class, VarIntWritable.class, VarLongWritable.class, IDMapReducer.class, VarIntWritable.class, VarLongWritable.class, SequenceFileOutputFormat.class); mapItems.getConfiguration().set(TOKEN_POS, String.valueOf(TasteHadoopUtils.ITEM_ID_POS)); mapItems.waitForCompletion(true); } Job itemRatings = prepareJob(getInputPath(), pathToItemRatings(), TextInputFormat.class, ItemRatingVectorsMapper.class, IntWritable.class, VectorWritable.class, VectorSumReducer.class, IntWritable.class, VectorWritable.class, SequenceFileOutputFormat.class); itemRatings.setCombinerClass(VectorSumCombiner.class); itemRatings.getConfiguration().set(USES_LONG_IDS, String.valueOf(usesLongIDs)); boolean succeeded = itemRatings.waitForCompletion(true); if (!succeeded) { return -1; } Job userRatings = prepareJob(pathToItemRatings(), pathToUserRatings(), TransposeMapper.class, IntWritable.class, VectorWritable.class, MergeUserVectorsReducer.class, IntWritable.class, VectorWritable.class); userRatings.setCombinerClass(MergeVectorsCombiner.class); succeeded = userRatings.waitForCompletion(true); if (!succeeded) { return -1; } Job averageItemRatings = prepareJob(pathToItemRatings(), getTempPath("averageRatings"), AverageRatingMapper.class, IntWritable.class, VectorWritable.class, MergeVectorsReducer.class, IntWritable.class, VectorWritable.class); averageItemRatings.setCombinerClass(MergeVectorsCombiner.class); succeeded = averageItemRatings.waitForCompletion(true); if (!succeeded) { return -1; } Vector averageRatings = ALS.readFirstRow(getTempPath("averageRatings"), getConf()); int numItems = averageRatings.getNumNondefaultElements(); int numUsers = (int) userRatings.getCounters().findCounter(Stats.NUM_USERS).getValue(); log.info("Found {} users and {} items", numUsers, numItems); initializeM(averageRatings); for (int currentIteration = 0; currentIteration < numIterations; currentIteration++) { log.info("Recomputing U (iteration {}/{})", currentIteration, numIterations); runSolver(pathToUserRatings(), pathToU(currentIteration), pathToM(currentIteration - 1), currentIteration, "U", numItems); log.info("Recomputing M (iteration {}/{})", currentIteration, numIterations); runSolver(pathToItemRatings(), pathToM(currentIteration), pathToU(currentIteration), currentIteration, "M", numUsers); } return 0; }  static void main(String[] args); @Override int run(String[] args);  }
@Test public void exampleWithIDMapping() throws Exception { String[] preferencesWithLongIDs = { "5568227754922264005,-4758971626494767444,5.0", "5568227754922264005,3688396615879561990,5.0", "5568227754922264005,4594226737871995304,2.0", "550945997885173934,-4758971626494767444,2.0", "550945997885173934,4594226737871995304,3.0", "550945997885173934,706816485922781596,5.0", "2448095297482319463,3688396615879561990,5.0", "2448095297482319463,706816485922781596,3.0", "6839920411763636962,-4758971626494767444,3.0", "6839920411763636962,706816485922781596,5.0" }; writeLines(inputFile, preferencesWithLongIDs); ParallelALSFactorizationJob alsFactorization = new ParallelALSFactorizationJob(); alsFactorization.setConf(conf); int numFeatures = 3; int numIterations = 5; double lambda = 0.065; alsFactorization.run(new String[] { "--input", inputFile.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--tempDir", tmpDir.getAbsolutePath(), "--lambda", String.valueOf(lambda), "--numFeatures", String.valueOf(numFeatures), "--numIterations", String.valueOf(numIterations), "--numThreadsPerSolver", String.valueOf(1), "--usesLongIDs", String.valueOf(true) }); OpenIntLongHashMap userIDIndex = TasteHadoopUtils.readIDIndexMap(outputDir.getAbsolutePath() + "/userIDIndex/part-r-00000", conf); assertEquals(4, userIDIndex.size()); OpenIntLongHashMap itemIDIndex = TasteHadoopUtils.readIDIndexMap(outputDir.getAbsolutePath() + "/itemIDIndex/part-r-00000", conf); assertEquals(4, itemIDIndex.size()); OpenIntObjectHashMap<Vector> u = MathHelper.readMatrixRows(conf, new Path(outputDir.getAbsolutePath(), "U/part-m-00000")); OpenIntObjectHashMap<Vector> m = MathHelper.readMatrixRows(conf, new Path(outputDir.getAbsolutePath(), "M/part-m-00000")); assertEquals(4, u.size()); assertEquals(4, m.size()); RunningAverage avg = new FullRunningAverage(); for (String line : preferencesWithLongIDs) { String[] tokens = TasteHadoopUtils.splitPrefTokens(line); long userID = Long.parseLong(tokens[TasteHadoopUtils.USER_ID_POS]); long itemID = Long.parseLong(tokens[TasteHadoopUtils.ITEM_ID_POS]); double rating = Double.parseDouble(tokens[2]); Vector userFeatures = u.get(TasteHadoopUtils.idToIndex(userID)); Vector itemFeatures = m.get(TasteHadoopUtils.idToIndex(itemID)); double estimate = userFeatures.dot(itemFeatures); double err = rating - estimate; avg.addDatum(err * err); } double rmse = Math.sqrt(avg.getAverage()); log.info("RMSE: {}", rmse); assertTrue(rmse < 0.2); }
@Test public void recommenderJobWithIDMapping() throws Exception { String[] preferencesWithLongIDs = { "5568227754922264005,-4758971626494767444,5.0", "5568227754922264005,3688396615879561990,5.0", "5568227754922264005,4594226737871995304,2.0", "550945997885173934,-4758971626494767444,2.0", "550945997885173934,4594226737871995304,3.0", "550945997885173934,706816485922781596,5.0", "2448095297482319463,3688396615879561990,5.0", "2448095297482319463,706816485922781596,3.0", "6839920411763636962,-4758971626494767444,3.0", "6839920411763636962,706816485922781596,5.0" }; writeLines(inputFile, preferencesWithLongIDs); ParallelALSFactorizationJob alsFactorization = new ParallelALSFactorizationJob(); alsFactorization.setConf(conf); int numFeatures = 3; int numIterations = 5; double lambda = 0.065; Configuration conf = getConfiguration(); int success = ToolRunner.run(alsFactorization, new String[] { "-Dhadoop.tmp.dir=" + conf.get("hadoop.tmp.dir"), "--input", inputFile.getAbsolutePath(), "--output", intermediateDir.getAbsolutePath(), "--tempDir", tmpDir.getAbsolutePath(), "--lambda", String.valueOf(lambda), "--numFeatures", String.valueOf(numFeatures), "--numIterations", String.valueOf(numIterations), "--numThreadsPerSolver", String.valueOf(1), "--usesLongIDs", String.valueOf(true) }); assertEquals(0, success); SharingMapper.reset(); RecommenderJob recommender = new RecommenderJob(); success = ToolRunner.run(recommender, new String[] { "-Dhadoop.tmp.dir=" + conf.get("hadoop.tmp.dir"), "--input", intermediateDir.getAbsolutePath() + "/userRatings/", "--userFeatures", intermediateDir.getAbsolutePath() + "/U/", "--itemFeatures", intermediateDir.getAbsolutePath() + "/M/", "--numRecommendations", String.valueOf(2), "--maxRating", String.valueOf(5.0), "--numThreads", String.valueOf(2), "--usesLongIDs", String.valueOf(true), "--userIDIndex", intermediateDir.getAbsolutePath() + "/userIDIndex/", "--itemIDIndex", intermediateDir.getAbsolutePath() + "/itemIDIndex/", "--output", outputDir.getAbsolutePath() }); assertEquals(0, success); }