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(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); if (getConf() == null) { setConf(new Configuration()); } String outType = getOption(OUT_TYPE_KEY); if (outType == null) { outType = "pw"; } run(getConf(), input, seeds, output, measure, outType); 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 final String SEEDS; static final String SEEDS_PATH_KEY; static final String DISTANCE_MEASURE_KEY; static final String OUT_TYPE_KEY; }
@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 = new Configuration(); 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(new Configuration(), new VectorDistanceSimilarityJob(), args); int expect = SEEDS.length * REFERENCE.length; DummyOutputCollector<StringTuple, DoubleWritable> collector = new DummyOutputCollector<StringTuple, DoubleWritable>(); for (Pair<StringTuple, DoubleWritable> record : new SequenceFileIterable<StringTuple, DoubleWritable>( new Path(output, "part-m-00000"), conf)) { collector.collect(record.getFirst(), record.getSecond()); } assertEquals(expect, collector.getData().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 = new Configuration(); 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(new Configuration(), 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()); } }