PredictionJob extends AbstractJob { @Override public int run(String[] args) throws Exception { addOption("pairs", "p", "path containing the test ratings, each line must be: userID,itemID", true); addOption("userFeatures", "u", "path to the user feature matrix", true); addOption("itemFeatures", "i", "path to the item feature matrix", true); addOutputOption(); Map<String,String> parsedArgs = parseArguments(args); if (parsedArgs == null) { return -1; } Path pairs = new Path(parsedArgs.get("--pairs")); Path userFeatures = new Path(parsedArgs.get("--userFeatures")); Path itemFeatures = new Path(parsedArgs.get("--itemFeatures")); Path tempDirPath = new Path(parsedArgs.get("--tempDir")); Path convertedPairs = new Path(tempDirPath, "convertedPairs"); Path convertedUserFeatures = new Path(tempDirPath, "convertedUserFeatures"); Path convertedItemFeatures = new Path(tempDirPath, "convertedItemFeatures"); Path pairsJoinedWithItemFeatures = new Path(tempDirPath, "pairsJoinedWithItemFeatures"); Job convertPairs = prepareJob(pairs, convertedPairs, TextInputFormat.class, PairsMapper.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, Reducer.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, SequenceFileOutputFormat.class); convertPairs.waitForCompletion(true); Job convertUserFeatures = prepareJob(userFeatures, convertedUserFeatures, SequenceFileInputFormat.class, FeaturesMapper.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, Reducer.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, SequenceFileOutputFormat.class); convertUserFeatures.waitForCompletion(true); Job convertItemFeatures = prepareJob(itemFeatures, convertedItemFeatures, SequenceFileInputFormat.class, FeaturesMapper.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, Reducer.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, SequenceFileOutputFormat.class); convertItemFeatures.waitForCompletion(true); Job joinPairsWithItemFeatures = prepareJob(new Path(convertedPairs + "," + convertedItemFeatures), pairsJoinedWithItemFeatures, SequenceFileInputFormat.class, Mapper.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, JoinProbesWithItemFeaturesReducer.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, SequenceFileOutputFormat.class); joinPairsWithItemFeatures.setPartitionerClass(HashPartitioner.class); joinPairsWithItemFeatures.setGroupingComparatorClass(TaggedVarIntWritable.GroupingComparator.class); joinPairsWithItemFeatures.waitForCompletion(true); Job predictRatings = prepareJob(new Path(pairsJoinedWithItemFeatures + "," + convertedUserFeatures), getOutputPath(), SequenceFileInputFormat.class, Mapper.class, TaggedVarIntWritable.class, VectorWithIndexWritable.class, PredictRatingReducer.class, Text.class, NullWritable.class, TextOutputFormat.class); predictRatings.setPartitionerClass(HashPartitioner.class); predictRatings.setGroupingComparatorClass(TaggedVarIntWritable.GroupingComparator.class); predictRatings.waitForCompletion(true); return 0; }  static void main(String[] args); @Override int run(String[] args);  }
@Test public void smallIntegration() throws Exception { File pairs = getTestTempFile("pairs.txt"); File userFeatures = getTestTempFile("userFeatures.seq"); File itemFeatures = getTestTempFile("itemFeatures.seq"); File tempDir = getTestTempDir("temp"); File outputDir = getTestTempDir("out"); outputDir.delete(); Configuration conf = new Configuration(); Path inputPath = new Path(pairs.getAbsolutePath()); FileSystem fs = FileSystem.get(inputPath.toUri(), conf); MathHelper.writeEntries(new double[][]{ new double[] { 1.5, -2, 0.3 }, new double[] { -0.7, 2, 0.6 }, new double[] { -1, 2.5, 3 } }, fs, conf, new Path(userFeatures.getAbsolutePath())); MathHelper.writeEntries(new double [][] { new double[] { 2.3, 0.5, 0 }, new double[] { 4.7, -1, 0.2 }, new double[] { 0.6, 2, 1.3 } }, fs, conf, new Path(itemFeatures.getAbsolutePath())); writeLines(pairs, "0,0", "2,1", "1,0"); PredictionJob predictor = new PredictionJob(); predictor.setConf(conf); predictor.run(new String[] { "--output", outputDir.getAbsolutePath(), "--pairs", pairs.getAbsolutePath(), "--userFeatures", userFeatures.getAbsolutePath(), "--itemFeatures", itemFeatures.getAbsolutePath(), "--tempDir", tempDir.getAbsolutePath() }); FileDataModel dataModel = new FileDataModel(new File(outputDir, "part-r-00000")); assertEquals(3, dataModel.getNumUsers()); assertEquals(2, dataModel.getNumItems()); assertEquals(2.45f, dataModel.getPreferenceValue(0, 0), EPSILON); assertEquals(-6.6f, dataModel.getPreferenceValue(2, 1), EPSILON); assertEquals(-0.61f, dataModel.getPreferenceValue(1, 0), EPSILON); }