ItemSimilarityJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption("similarityClassname", "s", "Name of distributed similarity measures class to instantiate, " + "alternatively use one of the predefined similarities (" + VectorSimilarityMeasures.list() + ')'); addOption("maxSimilaritiesPerItem", "m", "try to cap the number of similar items per item to this number " + "(default: " + DEFAULT_MAX_SIMILAR_ITEMS_PER_ITEM + ')', String.valueOf(DEFAULT_MAX_SIMILAR_ITEMS_PER_ITEM)); addOption("maxPrefs", "mppu", "max number of preferences to consider per user or item, " + "users or items with more preferences will be sampled down (default: " + DEFAULT_MAX_PREFS + ')', String.valueOf(DEFAULT_MAX_PREFS)); addOption("minPrefsPerUser", "mp", "ignore users with less preferences than this " + "(default: " + DEFAULT_MIN_PREFS_PER_USER + ')', String.valueOf(DEFAULT_MIN_PREFS_PER_USER)); addOption("booleanData", "b", "Treat input as without pref values", String.valueOf(Boolean.FALSE)); addOption("threshold", "tr", "discard item pairs with a similarity value below this", false); addOption("randomSeed", null, "use this seed for sampling", false); Map<String,List<String>> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } String similarityClassName = getOption("similarityClassname"); int maxSimilarItemsPerItem = Integer.parseInt(getOption("maxSimilaritiesPerItem")); int maxPrefs = Integer.parseInt(getOption("maxPrefs")); int minPrefsPerUser = Integer.parseInt(getOption("minPrefsPerUser")); boolean booleanData = Boolean.valueOf(getOption("booleanData")); double threshold = hasOption("threshold") ? Double.parseDouble(getOption("threshold")) : RowSimilarityJob.NO_THRESHOLD; long randomSeed = hasOption("randomSeed") ? Long.parseLong(getOption("randomSeed")) : RowSimilarityJob.NO_FIXED_RANDOM_SEED; Path similarityMatrixPath = getTempPath("similarityMatrix"); Path prepPath = getTempPath("prepareRatingMatrix"); AtomicInteger currentPhase = new AtomicInteger(); if (shouldRunNextPhase(parsedArgs, currentPhase)) { ToolRunner.run(getConf(), new PreparePreferenceMatrixJob(), new String[] { "--input", getInputPath().toString(), "--output", prepPath.toString(), "--minPrefsPerUser", String.valueOf(minPrefsPerUser), "--booleanData", String.valueOf(booleanData), "--tempDir", getTempPath().toString(), }); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { int numberOfUsers = HadoopUtil.readInt(new Path(prepPath, PreparePreferenceMatrixJob.NUM_USERS), 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, "--maxObservationsPerRow", String.valueOf(maxPrefs), "--maxObservationsPerColumn", String.valueOf(maxPrefs), "--maxSimilaritiesPerRow", String.valueOf(maxSimilarItemsPerItem), "--excludeSelfSimilarity", String.valueOf(Boolean.TRUE), "--threshold", String.valueOf(threshold), "--randomSeed", String.valueOf(randomSeed), "--tempDir", getTempPath().toString(), }); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job mostSimilarItems = prepareJob(similarityMatrixPath, getOutputPath(), SequenceFileInputFormat.class, MostSimilarItemPairsMapper.class, EntityEntityWritable.class, DoubleWritable.class, MostSimilarItemPairsReducer.class, EntityEntityWritable.class, DoubleWritable.class, TextOutputFormat.class); Configuration mostSimilarItemsConf = mostSimilarItems.getConfiguration(); mostSimilarItemsConf.set(ITEM_ID_INDEX_PATH_STR, new Path(prepPath, PreparePreferenceMatrixJob.ITEMID_INDEX).toString()); mostSimilarItemsConf.setInt(MAX_SIMILARITIES_PER_ITEM, maxSimilarItemsPerItem); boolean succeeded = mostSimilarItems.waitForCompletion(true); if (!succeeded) { return -1; } } return 0; }  static void main(String[] args); @Override int run(String[] args); static final String ITEM_ID_INDEX_PATH_STR; static final String MAX_SIMILARITIES_PER_ITEM; }
@Test public void testCompleteJob() throws Exception { File inputFile = getTestTempFile("prefs.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); writeLines(inputFile, "2,1,1", "1,2,1", "3,4,1", "1,3,2", "2,3,1"); ItemSimilarityJob similarityJob = new ItemSimilarityJob(); Configuration conf = getConfiguration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); similarityJob.setConf(conf); similarityJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname", CosineSimilarity.class.getName() }); File outPart = outputDir.listFiles(new FilenameFilter() { @Override public boolean accept(File dir, String name) { return name.startsWith("part-"); } })[0]; BufferedReader reader = Files.newReader(outPart, Charsets.UTF_8); String line; int currentLine = 1; while ( (line = reader.readLine()) != null) { String[] tokens = TAB.split(line); long itemAID = Long.parseLong(tokens[0]); long itemBID = Long.parseLong(tokens[1]); double similarity = Double.parseDouble(tokens[2]); if (currentLine == 1) { assertEquals(1L, itemAID); assertEquals(3L, itemBID); assertEquals(0.45, similarity, 0.01); } if (currentLine == 2) { assertEquals(2L, itemAID); assertEquals(3L, itemBID); assertEquals(0.89, similarity, 0.01); } currentLine++; } int linesWritten = currentLine-1; assertEquals(2, linesWritten); }
@Test public void testMaxSimilaritiesPerItem() throws Exception { File inputFile = getTestTempFile("prefsForMaxSimilarities.txt"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); writeLines(inputFile, "1,1,1", "1,3,1", "2,2,1", "2,3,1", "3,1,1", "3,2,1", "4,1,1", "4,2,1", "4,3,1", "5,2,1", "6,1,1", "6,2,1"); ItemSimilarityJob similarityJob = new ItemSimilarityJob(); Configuration conf = getConfiguration(); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); similarityJob.setConf(conf); similarityJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--maxSimilaritiesPerItem", "1" }); File outPart = outputDir.listFiles(new FilenameFilter() { @Override public boolean accept(File dir, String name) { return name.startsWith("part-"); } })[0]; BufferedReader reader = Files.newReader(outPart, Charsets.UTF_8); String line; int currentLine = 1; while ((line = reader.readLine()) != null) { String[] tokens = TAB.split(line); long itemAID = Long.parseLong(tokens[0]); long itemBID = Long.parseLong(tokens[1]); double similarity = Double.parseDouble(tokens[2]); if (currentLine == 1) { assertEquals(1L, itemAID); assertEquals(2L, itemBID); assertEquals(0.5, similarity, 0.0001); } if (currentLine == 2) { assertEquals(1L, itemAID); assertEquals(3L, itemBID); assertEquals(0.4, similarity, 0.0001); } currentLine++; } int linesWritten = currentLine - 1; assertEquals(2, linesWritten); }