PFPGrowth { public static void startParallelFPGrowth(Parameters params, Configuration conf) throws IOException, InterruptedException, ClassNotFoundException { conf.set(PFP_PARAMETERS, params.toString()); conf.set("mapred.compress.map.output", "true"); conf.set("mapred.output.compression.type", "BLOCK"); Path input = new Path(params.get(INPUT)); Job job = new Job(conf, "PFP Growth Driver running over input" + input); job.setJarByClass(PFPGrowth.class); job.setMapOutputKeyClass(IntWritable.class); job.setMapOutputValueClass(TransactionTree.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TopKStringPatterns.class); FileInputFormat.addInputPath(job, input); Path outPath = new Path(params.get(OUTPUT), FPGROWTH); FileOutputFormat.setOutputPath(job, outPath); HadoopUtil.delete(conf, outPath); job.setInputFormatClass(TextInputFormat.class); job.setMapperClass(ParallelFPGrowthMapper.class); job.setCombinerClass(ParallelFPGrowthCombiner.class); job.setReducerClass(ParallelFPGrowthReducer.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); job.waitForCompletion(true); } private  PFPGrowth(); static List<Pair<String,Long>> readFList(Configuration conf); static void saveFList(Iterable<Pair<String,Long>> flist, Parameters params, Configuration conf); static List<Pair<String,Long>> readFList(Parameters params); static final int getGroup(int itemId, int maxPerGroup); static final IntArrayList getGroupMembers(int groupId, 
                                                   int maxPerGroup, 
                                                   int numFeatures); static List<Pair<String,TopKStringPatterns>> readFrequentPattern(Parameters params); static void runPFPGrowth(Parameters params); static void startAggregating(Parameters params, Configuration conf); static void startParallelCounting(Parameters params, Configuration conf); static void startParallelFPGrowth(Parameters params, Configuration conf); static final String ENCODING; static final String F_LIST; static final String G_LIST; static final String NUM_GROUPS; static final int NUM_GROUPS_DEFAULT; static final String MAX_PER_GROUP; static final String OUTPUT; static final String MIN_SUPPORT; static final String MAX_HEAPSIZE; static final String INPUT; static final String PFP_PARAMETERS; static final String FILE_PATTERN; static final String FPGROWTH; static final String FREQUENT_PATTERNS; static final String PARALLEL_COUNTING; static final String SPLIT_PATTERN; static final String USE_FPG2; static final Pattern SPLITTER; }
@Test public void testStartParallelFPGrowth() throws Exception { Configuration conf = new Configuration(); PFPGrowth.runPFPGrowth(params); List<Pair<String,TopKStringPatterns>> frequentPatterns = PFPGrowth.readFrequentPattern(params); assertEquals("[(A,([A],5), ([D, A],4), ([B, A],4), ([A, E],4)), " + "(B,([B],6), ([B, D],4), ([B, A],4), ([B, D, A],3)), " + "(C,([B, C],3)), " + "(D,([D],6), ([D, A],4), ([B, D],4), ([D, A, E],3)), " + "(E,([A, E],4), ([D, A, E],3), ([B, A, E],3))]", frequentPatterns.toString()); }