RowSimilarityJob extends AbstractJob { @Override public int run(String[] args) throws IOException, ClassNotFoundException, InterruptedException { addInputOption(); addOutputOption(); addOption("numberOfColumns", "r", "Number of columns in the input matrix"); addOption("similarityClassname", "s", "Name of distributed similarity class to instantiate, alternatively use " + "one of the predefined similarities (" + SimilarityType.listEnumNames() + ')'); addOption("maxSimilaritiesPerRow", "m", "Number of maximum similarities per row (default: " + DEFAULT_MAX_SIMILARITIES_PER_ROW + ')', String.valueOf(DEFAULT_MAX_SIMILARITIES_PER_ROW)); Map<String,String> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } int numberOfColumns = Integer.parseInt(parsedArgs.get("--numberOfColumns")); String similarityClassnameArg = parsedArgs.get("--similarityClassname"); String distributedSimilarityClassname; try { distributedSimilarityClassname = SimilarityType.valueOf(similarityClassnameArg).getSimilarityImplementationClassName(); } catch (IllegalArgumentException iae) { distributedSimilarityClassname = similarityClassnameArg; } int maxSimilaritiesPerRow = Integer.parseInt(parsedArgs.get("--maxSimilaritiesPerRow")); Path inputPath = getInputPath(); Path outputPath = getOutputPath(); Path tempDirPath = new Path(parsedArgs.get("--tempDir")); Path weightsPath = new Path(tempDirPath, "weights"); Path pairwiseSimilarityPath = new Path(tempDirPath, "pairwiseSimilarity"); AtomicInteger currentPhase = new AtomicInteger(); if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job weights = prepareJob(inputPath, weightsPath, SequenceFileInputFormat.class, RowWeightMapper.class, VarIntWritable.class, WeightedOccurrence.class, WeightedOccurrencesPerColumnReducer.class, VarIntWritable.class, WeightedOccurrenceArray.class, SequenceFileOutputFormat.class); weights.getConfiguration().set(DISTRIBUTED_SIMILARITY_CLASSNAME, distributedSimilarityClassname); weights.waitForCompletion(true); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job pairwiseSimilarity = prepareJob(weightsPath, pairwiseSimilarityPath, SequenceFileInputFormat.class, CooccurrencesMapper.class, WeightedRowPair.class, Cooccurrence.class, SimilarityReducer.class, SimilarityMatrixEntryKey.class, DistributedRowMatrix.MatrixEntryWritable.class, SequenceFileOutputFormat.class); Configuration pairwiseConf = pairwiseSimilarity.getConfiguration(); pairwiseConf.set(DISTRIBUTED_SIMILARITY_CLASSNAME, distributedSimilarityClassname); pairwiseConf.setInt(NUMBER_OF_COLUMNS, numberOfColumns); pairwiseSimilarity.waitForCompletion(true); } if (shouldRunNextPhase(parsedArgs, currentPhase)) { Job asMatrix = prepareJob(pairwiseSimilarityPath, outputPath, SequenceFileInputFormat.class, Mapper.class, SimilarityMatrixEntryKey.class, DistributedRowMatrix.MatrixEntryWritable.class, EntriesToVectorsReducer.class, IntWritable.class, VectorWritable.class, SequenceFileOutputFormat.class); asMatrix.setPartitionerClass(HashPartitioner.class); asMatrix.setGroupingComparatorClass(SimilarityMatrixEntryKey.SimilarityMatrixEntryKeyGroupingComparator.class); asMatrix.getConfiguration().setInt(MAX_SIMILARITIES_PER_ROW, maxSimilaritiesPerRow); asMatrix.waitForCompletion(true); } return 0; }  static void main(String[] args); @Override int run(String[] args); static final String DISTRIBUTED_SIMILARITY_CLASSNAME; static final String NUMBER_OF_COLUMNS; static final String MAX_SIMILARITIES_PER_ROW; }
@Test public void testSmallSampleMatrix() throws Exception { File inputFile = getTestTempFile("rows"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Configuration conf = new Configuration(); Path inputPath = new Path(inputFile.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeEntries(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); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); RowSimilarityJob rowSimilarityJob = new RowSimilarityJob(); rowSimilarityJob.setConf(conf); rowSimilarityJob.run(new String[] { "--numberOfColumns", "3", "--similarityClassname", DistributedTanimotoCoefficientVectorSimilarity.class.getName(), "--tempDir", tmpDir.getAbsolutePath() }); Matrix similarityMatrix = MathHelper.readEntries(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.6666, similarityMatrix.get(0, 1), 0.0001); assertEquals(0.6666, similarityMatrix.get(1, 0), 0.0001); }
@Test public void testLimitEntriesInSimilarityMatrix() throws Exception { File inputFile = getTestTempFile("rows"); File outputDir = getTestTempDir("output"); outputDir.delete(); File tmpDir = getTestTempDir("tmp"); Configuration conf = new Configuration(); Path inputPath = new Path(inputFile.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeEntries(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); conf.set("mapred.input.dir", inputFile.getAbsolutePath()); conf.set("mapred.output.dir", outputDir.getAbsolutePath()); conf.setBoolean("mapred.output.compress", false); RowSimilarityJob rowSimilarityJob = new RowSimilarityJob(); rowSimilarityJob.setConf(conf); rowSimilarityJob.run(new String[] { "--numberOfColumns", "3", "--maxSimilaritiesPerRow", "1", "--similarityClassname", DistributedTanimotoCoefficientExcludeIdentityVectorSimilarity.class.getName(), "--tempDir", tmpDir.getAbsolutePath() }); Matrix similarityMatrix = MathHelper.readEntries(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); }