VectorDistanceSimilarityJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addInputOption(); addOutputOption(); addOption(DefaultOptionCreator.distanceMeasureOption().create()); addOption(SEEDS, "s", "The set of vectors to compute distances against. Must fit in memory on the mapper"); addOption(MAX_DISTANCE, "mx", "set an upper-bound on distance (double) such that any pair of vectors with a" + " distance greater than this value is ignored in the output. Ignored for non pairwise output!"); addOption(DefaultOptionCreator.overwriteOption().create()); addOption(OUT_TYPE_KEY, "ot", "[pw|v] -- Define the output style: pairwise, the default, (pw) or vector (v). " + "Pairwise is a tuple of <seed, other, distance>, vector is <other, <Vector of size the number of seeds>>.", "pw"); if (parseArguments(args) == null) { return -1; } Path input = getInputPath(); Path output = getOutputPath(); Path seeds = new Path(getOption(SEEDS)); String measureClass = getOption(DefaultOptionCreator.DISTANCE_MEASURE_OPTION); if (measureClass == null) { measureClass = SquaredEuclideanDistanceMeasure.class.getName(); } if (hasOption(DefaultOptionCreator.OVERWRITE_OPTION)) { HadoopUtil.delete(getConf(), output); } DistanceMeasure measure = ClassUtils.instantiateAs(measureClass, DistanceMeasure.class); String outType = getOption(OUT_TYPE_KEY, "pw"); Double maxDistance = null; if ("pw".equals(outType)) { String maxDistanceArg = getOption(MAX_DISTANCE); if (maxDistanceArg != null) { maxDistance = Double.parseDouble(maxDistanceArg); Preconditions.checkArgument(maxDistance > 0.0d, "value for " + MAX_DISTANCE + " must be greater than zero"); } } run(getConf(), input, seeds, output, measure, outType, maxDistance); return 0; }  static void main(String[] args); @Override int run(String[] args); static void run(Configuration conf,
                         Path input,
                         Path seeds,
                         Path output,
                         DistanceMeasure measure, String outType); static void run(Configuration conf,
          Path input,
          Path seeds,
          Path output,
          DistanceMeasure measure, String outType, Double maxDistance); static final String SEEDS; static final String SEEDS_PATH_KEY; static final String DISTANCE_MEASURE_KEY; static final String OUT_TYPE_KEY; static final String MAX_DISTANCE; }
@Test public void testRun() throws Exception { Path input = getTestTempDirPath("input"); Path output = getTestTempDirPath("output"); Path seedsPath = getTestTempDirPath("seeds"); List<VectorWritable> points = getPointsWritable(REFERENCE); List<VectorWritable> seeds = getPointsWritable(SEEDS); Configuration conf = getConfiguration(); ClusteringTestUtils.writePointsToFile(points, true, new Path(input, "file1"), fs, conf); ClusteringTestUtils.writePointsToFile(seeds, true, new Path(seedsPath, "part-seeds"), fs, conf); String[] args = { optKey(DefaultOptionCreator.INPUT_OPTION), input.toString(), optKey(VectorDistanceSimilarityJob.SEEDS), seedsPath.toString(), optKey(DefaultOptionCreator.OUTPUT_OPTION), output.toString(), optKey(DefaultOptionCreator.DISTANCE_MEASURE_OPTION), EuclideanDistanceMeasure.class.getName() }; ToolRunner.run(getConfiguration(), new VectorDistanceSimilarityJob(), args); int expectedOutputSize = SEEDS.length * REFERENCE.length; int outputSize = Iterables.size(new SequenceFileIterable<StringTuple, DoubleWritable>(new Path(output, "part-m-00000"), conf)); assertEquals(expectedOutputSize, outputSize); }
@Test public void testMaxDistance() throws Exception { Path input = getTestTempDirPath("input"); Path output = getTestTempDirPath("output"); Path seedsPath = getTestTempDirPath("seeds"); List<VectorWritable> points = getPointsWritable(REFERENCE); List<VectorWritable> seeds = getPointsWritable(SEEDS); Configuration conf = getConfiguration(); ClusteringTestUtils.writePointsToFile(points, true, new Path(input, "file1"), fs, conf); ClusteringTestUtils.writePointsToFile(seeds, true, new Path(seedsPath, "part-seeds"), fs, conf); double maxDistance = 10; String[] args = { optKey(DefaultOptionCreator.INPUT_OPTION), input.toString(), optKey(VectorDistanceSimilarityJob.SEEDS), seedsPath.toString(), optKey(DefaultOptionCreator.OUTPUT_OPTION), output.toString(), optKey(DefaultOptionCreator.DISTANCE_MEASURE_OPTION), EuclideanDistanceMeasure.class.getName(), optKey(VectorDistanceSimilarityJob.MAX_DISTANCE), String.valueOf(maxDistance) }; ToolRunner.run(getConfiguration(), new VectorDistanceSimilarityJob(), args); int outputSize = 0; for (Pair<StringTuple, DoubleWritable> record : new SequenceFileIterable<StringTuple, DoubleWritable>( new Path(output, "part-m-00000"), conf)) { outputSize++; assertTrue(record.getSecond().get() <= maxDistance); } assertEquals(14, outputSize); }
@Test public void testRunInverted() throws Exception { Path input = getTestTempDirPath("input"); Path output = getTestTempDirPath("output"); Path seedsPath = getTestTempDirPath("seeds"); List<VectorWritable> points = getPointsWritable(REFERENCE); List<VectorWritable> seeds = getPointsWritable(SEEDS); Configuration conf = getConfiguration(); ClusteringTestUtils.writePointsToFile(points, true, new Path(input, "file1"), fs, conf); ClusteringTestUtils.writePointsToFile(seeds, true, new Path(seedsPath, "part-seeds"), fs, conf); String[] args = {optKey(DefaultOptionCreator.INPUT_OPTION), input.toString(), optKey(VectorDistanceSimilarityJob.SEEDS), seedsPath.toString(), optKey(DefaultOptionCreator.OUTPUT_OPTION), output.toString(), optKey(DefaultOptionCreator.DISTANCE_MEASURE_OPTION), EuclideanDistanceMeasure.class.getName(), optKey(VectorDistanceSimilarityJob.OUT_TYPE_KEY), "v" }; ToolRunner.run(getConfiguration(), new VectorDistanceSimilarityJob(), args); DummyOutputCollector<Text, VectorWritable> collector = new DummyOutputCollector<Text, VectorWritable>(); for (Pair<Text, VectorWritable> record : new SequenceFileIterable<Text, VectorWritable>( new Path(output, "part-m-00000"), conf)) { collector.collect(record.getFirst(), record.getSecond()); } assertEquals(REFERENCE.length, collector.getData().size()); for (Map.Entry<Text, List<VectorWritable>> entry : collector.getData().entrySet()) { assertEquals(SEEDS.length, entry.getValue().iterator().next().get().size()); } }
@Test public void testRunInverted() throws Exception { Path input = getTestTempDirPath("input"); Path output = getTestTempDirPath("output"); Path seedsPath = getTestTempDirPath("seeds"); List<VectorWritable> points = getPointsWritable(REFERENCE); List<VectorWritable> seeds = getPointsWritable(SEEDS); Configuration conf = getConfiguration(); ClusteringTestUtils.writePointsToFile(points, true, new Path(input, "file1"), fs, conf); ClusteringTestUtils.writePointsToFile(seeds, true, new Path(seedsPath, "part-seeds"), fs, conf); String[] args = {optKey(DefaultOptionCreator.INPUT_OPTION), input.toString(), optKey(VectorDistanceSimilarityJob.SEEDS), seedsPath.toString(), optKey(DefaultOptionCreator.OUTPUT_OPTION), output.toString(), optKey(DefaultOptionCreator.DISTANCE_MEASURE_OPTION), EuclideanDistanceMeasure.class.getName(), optKey(VectorDistanceSimilarityJob.OUT_TYPE_KEY), "v" }; ToolRunner.run(getConfiguration(), new VectorDistanceSimilarityJob(), args); DummyOutputCollector<Text, VectorWritable> collector = new DummyOutputCollector<>(); for (Pair<Text, VectorWritable> record : new SequenceFileIterable<Text, VectorWritable>( new Path(output, "part-m-00000"), conf)) { collector.collect(record.getFirst(), record.getSecond()); } assertEquals(REFERENCE.length, collector.getData().size()); for (Map.Entry<Text, List<VectorWritable>> entry : collector.getData().entrySet()) { assertEquals(SEEDS.length, entry.getValue().iterator().next().get().size()); } }