RecommenderJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption("numRecommendations", "n", "Number of recommendations per user", String.valueOf(AggregateAndRecommendReducer.DEFAULT_NUM_RECOMMENDATIONS)); addOption("usersFile", null, "File of users to recommend for", null); addOption("itemsFile", null, "File of items to recommend for", null); addOption("filterFile", "f", "File containing comma-separated userID,itemID pairs. Used to exclude the item from " + "the recommendations for that user (optional)", null); addOption("booleanData", "b", "Treat input as without pref values", Boolean.FALSE.toString()); addOption("maxPrefsPerUser", "mxp", "Maximum number of preferences considered per user in final recommendation phase", String.valueOf(UserVectorSplitterMapper.DEFAULT_MAX_PREFS_PER_USER_CONSIDERED)); addOption("minPrefsPerUser", "mp", "ignore users with less preferences than this in the similarity computation " + "(default: " + DEFAULT_MIN_PREFS_PER_USER + ')', String.valueOf(DEFAULT_MIN_PREFS_PER_USER)); addOption("maxSimilaritiesPerItem", "m", "Maximum number of similarities considered per item ", String.valueOf(DEFAULT_MAX_SIMILARITIES_PER_ITEM)); addOption("maxPrefsPerUserInItemSimilarity", "mppuiis", "max number of preferences to consider per user in the " + "item similarity computation phase, users with more preferences will be sampled down (default: " + DEFAULT_MAX_PREFS_PER_USER + ')', String.valueOf(DEFAULT_MAX_PREFS_PER_USER)); addOption("similarityClassname", "s", "Name of distributed similarity measures class to instantiate, " + "alternatively use one of the predefined similarities (" + VectorSimilarityMeasures.list() + ')', true); addOption("threshold", "tr", "discard item pairs with a similarity value below this", false); Map<String, String> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } Path outputPath = getOutputPath(); int numRecommendations = Integer.parseInt(parsedArgs.get("--numRecommendations")); String usersFile = parsedArgs.get("--usersFile"); String itemsFile = parsedArgs.get("--itemsFile"); String filterFile = parsedArgs.get("--filterFile"); boolean booleanData = Boolean.valueOf(parsedArgs.get("--booleanData")); int maxPrefsPerUser = Integer.parseInt(parsedArgs.get("--maxPrefsPerUser")); int minPrefsPerUser = Integer.parseInt(parsedArgs.get("--minPrefsPerUser")); int maxPrefsPerUserInItemSimilarity = Integer.parseInt(parsedArgs.get("--maxPrefsPerUserInItemSimilarity")); int maxSimilaritiesPerItem = Integer.parseInt(parsedArgs.get("--maxSimilaritiesPerItem")); String similarityClassname = parsedArgs.get("--similarityClassname"); double threshold = parsedArgs.containsKey("--threshold") ? Double.parseDouble(parsedArgs.get("--threshold")) : RowSimilarityJob.NO_THRESHOLD; Path prepPath = getTempPath("preparePreferenceMatrix"); Path similarityMatrixPath = getTempPath("similarityMatrix"); Path prePartialMultiplyPath1 = getTempPath("prePartialMultiply1"); Path prePartialMultiplyPath2 = getTempPath("prePartialMultiply2"); Path explicitFilterPath = getTempPath("explicitFilterPath"); Path partialMultiplyPath = getTempPath("partialMultiply"); AtomicInteger currentPhase = new AtomicInteger(); int numberOfUsers = -1; if (shouldRunNextPhase(parsedArgs, currentPhase)) { ToolRunner.run(getConf(), new PreparePreferenceMatrixJob(), new String[]{ "--input", getInputPath().toString(), "--output", prepPath.toString(), "--maxPrefsPerUser", String.valueOf(maxPrefsPerUserInItemSimilarity), "--minPrefsPerUser", String.valueOf(minPrefsPerUser), "--booleanData", String.valueOf(booleanData), "--tempDir", getTempPath().toString()}); numberOfUsers = HadoopUtil.readInt(new Path(prepPath, PreparePreferenceMatrixJob.NUM_USERS), getConf()); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { if (numberOfUsers == -1) { numberOfUsers = (int) HadoopUtil.countRecords(new Path(prepPath, PreparePreferenceMatrixJob.USER_VECTORS), PathType.LIST, null, getConf()); } ToolRunner.run(getConf(), new RowSimilarityJob(), new String[]{ "--input", new Path(prepPath, PreparePreferenceMatrixJob.RATING_MATRIX).toString(), "--output", similarityMatrixPath.toString(), "--numberOfColumns", String.valueOf(numberOfUsers), "--similarityClassname", similarityClassname, "--maxSimilaritiesPerRow", String.valueOf(maxSimilaritiesPerItem), "--excludeSelfSimilarity", String.valueOf(Boolean.TRUE), "--threshold", String.valueOf(threshold), "--tempDir", getTempPath().toString()}); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job prePartialMultiply1 = prepareJob( similarityMatrixPath, prePartialMultiplyPath1, SequenceFileInputFormat.class, SimilarityMatrixRowWrapperMapper.class, VarIntWritable.class, VectorOrPrefWritable.class, Reducer.class, VarIntWritable.class, VectorOrPrefWritable.class, SequenceFileOutputFormat.class); prePartialMultiply1.waitForCompletion(true); 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); prePartialMultiply2.waitForCompletion(true); 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); partialMultiply.waitForCompletion(true); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { if (filterFile != null) { Job itemFiltering = prepareJob(new Path(filterFile), explicitFilterPath, TextInputFormat.class, ItemFilterMapper.class, VarLongWritable.class, VarLongWritable.class, ItemFilterAsVectorAndPrefsReducer.class, VarIntWritable.class, VectorAndPrefsWritable.class, SequenceFileOutputFormat.class); itemFiltering.waitForCompletion(true); } String aggregateAndRecommendInput = partialMultiplyPath.toString(); if (filterFile != null) { aggregateAndRecommendInput += "," + explicitFilterPath; } Job aggregateAndRecommend = prepareJob( new Path(aggregateAndRecommendInput), outputPath, SequenceFileInputFormat.class, PartialMultiplyMapper.class, VarLongWritable.class, PrefAndSimilarityColumnWritable.class, AggregateAndRecommendReducer.class, VarLongWritable.class, RecommendedItemsWritable.class, TextOutputFormat.class); Configuration aggregateAndRecommendConf = aggregateAndRecommend.getConfiguration(); if (itemsFile != null) { aggregateAndRecommendConf.set(AggregateAndRecommendReducer.ITEMS_FILE, itemsFile); } if (filterFile != null) { setS3SafeCombinedInputPath(aggregateAndRecommend, getTempPath(), partialMultiplyPath, explicitFilterPath); } setIOSort(aggregateAndRecommend); aggregateAndRecommendConf.set(AggregateAndRecommendReducer.ITEMID_INDEX_PATH, new Path(prepPath, PreparePreferenceMatrixJob.ITEMID_INDEX).toString()); aggregateAndRecommendConf.setInt(AggregateAndRecommendReducer.NUM_RECOMMENDATIONS, numRecommendations); aggregateAndRecommendConf.setBoolean(BOOLEAN_DATA, booleanData); aggregateAndRecommend.waitForCompletion(true); } return 0; }  @Override int run(String[] args); static void main(String[] args); static final String BOOLEAN_DATA; }
@Test public void testCompleteJob() throws Exception { File inputFile = getTestTempFile("prefs.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); writeLines(inputFile, "1,1,5", "1,2,5", "1,3,2", "2,1,2", "2,3,3", "2,4,5", "3,2,5", "3,4,3", "4,1,3", "4,4,5"); RecommenderJob recommenderJob = new RecommenderJob(); Configuration conf = new Configuration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); recommenderJob.setConf(conf); recommenderJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--numRecommendations", "4" }); Map<Long,List<RecommendedItem>> recommendations = readRecommendations(new File(outputDir, "part-r-00000")); assertEquals(4, recommendations.size()); for (Entry<Long,List<RecommendedItem>> entry : recommendations.entrySet()) { long userID = entry.getKey(); List<RecommendedItem> items = entry.getValue(); assertNotNull(items); RecommendedItem item1 = items.get(0); if (userID == 1L) { assertEquals(1, items.size()); assertEquals(4L, item1.getItemID()); assertEquals(4.3, item1.getValue(), 0.05); } if (userID == 2L) { assertEquals(1, items.size()); assertEquals(2L, item1.getItemID()); assertEquals(3.3, item1.getValue(), 0.05); } if (userID == 3L) { assertEquals(2, items.size()); assertEquals(3L, item1.getItemID()); assertEquals(4.1, item1.getValue(), 0.05); RecommendedItem item2 = items.get(1); assertEquals(1L, item2.getItemID()); assertEquals(3.7, item2.getValue(), 0.05); } if (userID == 4L) { assertEquals(2, items.size()); assertEquals(2L, item1.getItemID()); assertEquals(4.0, item1.getValue(), 0.05); RecommendedItem item2 = items.get(1); assertEquals(3L, item2.getItemID()); assertEquals(3.5, item2.getValue(), 0.05); } } }
@Test public void testCompleteJobBoolean() throws Exception { File inputFile = getTestTempFile("prefs.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); File usersFile = getTestTempFile("users.txt"); writeLines(usersFile, "3"); writeLines(inputFile, "1,1", "1,2", "1,3", "2,1", "2,3", "2,4", "3,2", "3,4", "4,1", "4,4"); RecommenderJob recommenderJob = new RecommenderJob(); Configuration conf = new Configuration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); recommenderJob.setConf(conf); recommenderJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname", CooccurrenceCountSimilarity.class.getName(), "--booleanData", "true", "--usersFile", usersFile.getAbsolutePath() }); Map<Long,List<RecommendedItem>> recommendations = readRecommendations(new File(outputDir, "part-r-00000")); List<RecommendedItem> recommendedToCow = recommendations.get(3L); assertEquals(2, recommendedToCow.size()); RecommendedItem item1 = recommendedToCow.get(0); RecommendedItem item2 = recommendedToCow.get(1); assertEquals(1L, item1.getItemID()); assertEquals(3L, item2.getItemID()); assertEquals(3, item1.getValue(), 0.05); assertEquals(2, item2.getValue(), 0.05); }
@Test public void testCompleteJobWithFiltering() throws Exception { File inputFile = getTestTempFile("prefs.txt"); File userFile = getTestTempFile("users.txt"); File filterFile = getTestTempFile("filter.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); writeLines(inputFile, "1,1,5", "1,2,5", "1,3,2", "2,1,2", "2,3,3", "2,4,5", "3,2,5", "3,4,3", "4,1,3", "4,4,5"); writeLines(userFile, "4"); writeLines(filterFile, "4,2"); RecommenderJob recommenderJob = new RecommenderJob(); Configuration conf = new Configuration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); recommenderJob.setConf(conf); recommenderJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--numRecommendations", "1", "--usersFile", userFile.getAbsolutePath(), "--filterFile", filterFile.getAbsolutePath() }); Map<Long,List<RecommendedItem>> recommendations = readRecommendations(new File(outputDir, "part-r-00000")); assertEquals(1, recommendations.size()); assertTrue(recommendations.containsKey(4L)); assertEquals(1, recommendations.get(4L).size()); RecommendedItem recommendedItem = recommendations.get(4L).get(0); assertEquals(3L, recommendedItem.getItemID()); assertEquals(3.5, recommendedItem.getValue(), 0.05); }