RowSimilarityJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption("numberOfColumns", "r", "Number of columns in the input matrix", false); addOption("similarityClassname", "s", "Name of distributed similarity class to instantiate, alternatively use " + "one of the predefined similarities (" + VectorSimilarityMeasures.list() + ')'); addOption("maxSimilaritiesPerRow", "m", "Number of maximum similarities per row (default: " + DEFAULT_MAX_SIMILARITIES_PER_ROW + ')', String.valueOf(DEFAULT_MAX_SIMILARITIES_PER_ROW)); addOption("excludeSelfSimilarity", "ess", "compute similarity of rows to themselves?", String.valueOf(false)); addOption("threshold", "tr", "discard row pairs with a similarity value below this", false); addOption("maxObservationsPerRow", null, "sample rows down to this number of entries", String.valueOf(DEFAULT_MAX_OBSERVATIONS_PER_ROW)); addOption("maxObservationsPerColumn", null, "sample columns down to this number of entries", String.valueOf(DEFAULT_MAX_OBSERVATIONS_PER_COLUMN)); addOption("randomSeed", null, "use this seed for sampling", false); addOption(DefaultOptionCreator.overwriteOption().create()); Map<String,List<String>> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } int numberOfColumns; if (hasOption("numberOfColumns")) { numberOfColumns = Integer.parseInt(getOption("numberOfColumns")); } else { numberOfColumns = getDimensions(getInputPath()); } String similarityClassnameArg = getOption("similarityClassname"); String similarityClassname; try { similarityClassname = VectorSimilarityMeasures.valueOf(similarityClassnameArg).getClassname(); } catch (IllegalArgumentException iae) { similarityClassname = similarityClassnameArg; } if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) { HadoopUtil.delete(getConf(), getTempPath()); HadoopUtil.delete(getConf(), getOutputPath()); } int maxSimilaritiesPerRow = Integer.parseInt(getOption("maxSimilaritiesPerRow")); boolean excludeSelfSimilarity = Boolean.parseBoolean(getOption("excludeSelfSimilarity")); double threshold = hasOption("threshold") ? Double.parseDouble(getOption("threshold")) : NO_THRESHOLD; long randomSeed = hasOption("randomSeed") ? Long.parseLong(getOption("randomSeed")) : NO_FIXED_RANDOM_SEED; int maxObservationsPerRow = Integer.parseInt(getOption("maxObservationsPerRow")); int maxObservationsPerColumn = Integer.parseInt(getOption("maxObservationsPerColumn")); Path weightsPath = getTempPath("weights"); Path normsPath = getTempPath("norms.bin"); Path numNonZeroEntriesPath = getTempPath("numNonZeroEntries.bin"); Path maxValuesPath = getTempPath("maxValues.bin"); Path pairwiseSimilarityPath = getTempPath("pairwiseSimilarity"); Path observationsPerColumnPath = getTempPath("observationsPerColumn.bin"); AtomicInteger currentPhase = new AtomicInteger(); Job countObservations = prepareJob(getInputPath(), getTempPath("notUsed"), CountObservationsMapper.class, NullWritable.class, VectorWritable.class, SumObservationsReducer.class, NullWritable.class, VectorWritable.class); countObservations.setCombinerClass(VectorSumCombiner.class); countObservations.getConfiguration().set(OBSERVATIONS_PER_COLUMN_PATH, observationsPerColumnPath.toString()); countObservations.setNumReduceTasks(1); countObservations.waitForCompletion(true); if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job normsAndTranspose = prepareJob(getInputPath(), weightsPath, VectorNormMapper.class, IntWritable.class, VectorWritable.class, MergeVectorsReducer.class, IntWritable.class, VectorWritable.class); normsAndTranspose.setCombinerClass(MergeVectorsCombiner.class); Configuration normsAndTransposeConf = normsAndTranspose.getConfiguration(); normsAndTransposeConf.set(THRESHOLD, String.valueOf(threshold)); normsAndTransposeConf.set(NORMS_PATH, normsPath.toString()); normsAndTransposeConf.set(NUM_NON_ZERO_ENTRIES_PATH, numNonZeroEntriesPath.toString()); normsAndTransposeConf.set(MAXVALUES_PATH, maxValuesPath.toString()); normsAndTransposeConf.set(SIMILARITY_CLASSNAME, similarityClassname); normsAndTransposeConf.set(OBSERVATIONS_PER_COLUMN_PATH, observationsPerColumnPath.toString()); normsAndTransposeConf.set(MAX_OBSERVATIONS_PER_ROW, String.valueOf(maxObservationsPerRow)); normsAndTransposeConf.set(MAX_OBSERVATIONS_PER_COLUMN, String.valueOf(maxObservationsPerColumn)); normsAndTransposeConf.set(RANDOM_SEED, String.valueOf(randomSeed)); boolean succeeded = normsAndTranspose.waitForCompletion(true); if (!succeeded) { return -1; } } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job pairwiseSimilarity = prepareJob(weightsPath, pairwiseSimilarityPath, CooccurrencesMapper.class, IntWritable.class, VectorWritable.class, SimilarityReducer.class, IntWritable.class, VectorWritable.class); pairwiseSimilarity.setCombinerClass(VectorSumReducer.class); Configuration pairwiseConf = pairwiseSimilarity.getConfiguration(); pairwiseConf.set(THRESHOLD, String.valueOf(threshold)); pairwiseConf.set(NORMS_PATH, normsPath.toString()); pairwiseConf.set(NUM_NON_ZERO_ENTRIES_PATH, numNonZeroEntriesPath.toString()); pairwiseConf.set(MAXVALUES_PATH, maxValuesPath.toString()); pairwiseConf.set(SIMILARITY_CLASSNAME, similarityClassname); pairwiseConf.setInt(NUMBER_OF_COLUMNS, numberOfColumns); pairwiseConf.setBoolean(EXCLUDE_SELF_SIMILARITY, excludeSelfSimilarity); boolean succeeded = pairwiseSimilarity.waitForCompletion(true); if (!succeeded) { return -1; } } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job asMatrix = prepareJob(pairwiseSimilarityPath, getOutputPath(), UnsymmetrifyMapper.class, IntWritable.class, VectorWritable.class, MergeToTopKSimilaritiesReducer.class, IntWritable.class, VectorWritable.class); asMatrix.setCombinerClass(MergeToTopKSimilaritiesReducer.class); asMatrix.getConfiguration().setInt(MAX_SIMILARITIES_PER_ROW, maxSimilaritiesPerRow); boolean succeeded = asMatrix.waitForCompletion(true); if (!succeeded) { return -1; } } return 0; }  static void main(String[] args); @Override int run(String[] args); static final double NO_THRESHOLD; static final long NO_FIXED_RANDOM_SEED; }
@Test public void toyIntegration() throws Exception { File inputFile = getTestTempFile("rows"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Configuration conf = getConfiguration(); Path inputPath = new Path(inputFile.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeDistributedRowMatrix(new double[][] { new double[] { 1, 0, 1, 1, 0 }, new double[] { 0, 0, 1, 1, 0 }, new double[] { 0, 0, 0, 0, 1 } }, fs, conf, inputPath); RowSimilarityJob rowSimilarityJob = new RowSimilarityJob(); rowSimilarityJob.setConf(conf); rowSimilarityJob.run(new String[] { "--input", inputFile.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--numberOfColumns", String.valueOf(5), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--tempDir", tmpDir.getAbsolutePath() }); OpenIntIntHashMap observationsPerColumn = Vectors.readAsIntMap(new Path(tmpDir.getAbsolutePath(), "observationsPerColumn.bin"), conf); assertEquals(4, observationsPerColumn.size()); assertEquals(1, observationsPerColumn.get(0)); assertEquals(2, observationsPerColumn.get(2)); assertEquals(2, observationsPerColumn.get(3)); assertEquals(1, observationsPerColumn.get(4)); Matrix similarityMatrix = MathHelper.readMatrix(conf, new Path(outputDir.getAbsolutePath(), "part-r-00000"), 3, 3); assertNotNull(similarityMatrix); assertEquals(3, similarityMatrix.numCols()); assertEquals(3, similarityMatrix.numRows()); assertEquals(1.0, similarityMatrix.get(0, 0), EPSILON); assertEquals(1.0, similarityMatrix.get(1, 1), EPSILON); assertEquals(1.0, similarityMatrix.get(2, 2), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 0), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(0, 2), EPSILON); assertEquals(0.0, similarityMatrix.get(1, 2), EPSILON); assertEquals(0.666666, similarityMatrix.get(0, 1), EPSILON); assertEquals(0.666666, similarityMatrix.get(1, 0), EPSILON); }
@Test public void toyIntegrationMaxSimilaritiesPerRow() throws Exception { File inputFile = getTestTempFile("rows"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Configuration conf = getConfiguration(); Path inputPath = new Path(inputFile.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeDistributedRowMatrix(new double[][]{ new double[] { 1, 0, 1, 1, 0, 1 }, new double[] { 0, 1, 1, 1, 1, 1 }, new double[] { 1, 1, 0, 1, 0, 0 } }, fs, conf, inputPath); RowSimilarityJob rowSimilarityJob = new RowSimilarityJob(); rowSimilarityJob.setConf(conf); rowSimilarityJob.run(new String[] { "--input", inputFile.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--numberOfColumns", String.valueOf(6), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--maxSimilaritiesPerRow", String.valueOf(1), "--excludeSelfSimilarity", String.valueOf(true), "--tempDir", tmpDir.getAbsolutePath() }); Matrix similarityMatrix = MathHelper.readMatrix(conf, new Path(outputDir.getAbsolutePath(), "part-r-00000"), 3, 3); assertNotNull(similarityMatrix); assertEquals(3, similarityMatrix.numCols()); assertEquals(3, similarityMatrix.numRows()); assertEquals(0.0, similarityMatrix.get(0, 0), EPSILON); assertEquals(0.5, similarityMatrix.get(0, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(0, 2), EPSILON); assertEquals(0.5, similarityMatrix.get(1, 0), EPSILON); assertEquals(0.0, similarityMatrix.get(1, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(1, 2), EPSILON); assertEquals(0.4, similarityMatrix.get(2, 0), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 2), EPSILON); }
@Test public void toyIntegrationWithThreshold() throws Exception { File inputFile = getTestTempFile("rows"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Configuration conf = getConfiguration(); Path inputPath = new Path(inputFile.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeDistributedRowMatrix(new double[][]{ new double[] { 1, 0, 1, 1, 0, 1 }, new double[] { 0, 1, 1, 1, 1, 1 }, new double[] { 1, 1, 0, 1, 0, 0 } }, fs, conf, inputPath); RowSimilarityJob rowSimilarityJob = new RowSimilarityJob(); rowSimilarityJob.setConf(conf); rowSimilarityJob.run(new String[] { "--input", inputFile.getAbsolutePath(), "--output", outputDir.getAbsolutePath(), "--numberOfColumns", String.valueOf(6), "--similarityClassname", TanimotoCoefficientSimilarity.class.getName(), "--excludeSelfSimilarity", String.valueOf(true), "--threshold", String.valueOf(0.5), "--tempDir", tmpDir.getAbsolutePath() }); Matrix similarityMatrix = MathHelper.readMatrix(conf, new Path(outputDir.getAbsolutePath(), "part-r-00000"), 3, 3); assertNotNull(similarityMatrix); assertEquals(3, similarityMatrix.numCols()); assertEquals(3, similarityMatrix.numRows()); assertEquals(0.0, similarityMatrix.get(0, 0), EPSILON); assertEquals(0.5, similarityMatrix.get(0, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(0, 2), EPSILON); assertEquals(0.5, similarityMatrix.get(1, 0), EPSILON); assertEquals(0.0, similarityMatrix.get(1, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(1, 2), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 0), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 1), EPSILON); assertEquals(0.0, similarityMatrix.get(2, 2), EPSILON); }