PFPGrowth { public static void startParallelFPGrowth(Parameters params) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); 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(OUTPUT), SORTED_OUTPUT); Job job = new Job(conf, "PFP Growth Driver running over input" + input); job.setJarByClass(PFPGrowth.class); job.setMapOutputKeyClass(LongWritable.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(SequenceFileInputFormat.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>> deserializeList(Parameters params,
                                                        String key,
                                                        Configuration conf); static Map<String,Long> deserializeMap(Parameters params, String key, Configuration conf); static List<Pair<String,Long>> readFList(Parameters params); static List<Pair<String,TopKStringPatterns>> readFrequentPattern(Parameters params); static void runPFPGrowth(Parameters params); static void startAggregating(Parameters params); static void startGroupingItems(Parameters params); static void startParallelCounting(Parameters params); static void startTransactionSorting(Parameters params); static void startParallelFPGrowth(Parameters params); static final String ENCODING; static final String F_LIST; static final String G_LIST; static final String NUM_GROUPS; 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 SORTED_OUTPUT; static final String SPLIT_PATTERN; static final Pattern SPLITTER; }
@Test public void testStartParallelFPGrowth() throws Exception { log.info("Starting Parallel Counting Test: {}", params.get(PFPGrowth.MAX_HEAPSIZE)); PFPGrowth.startParallelCounting(params); log.info("Reading fList Test: {}", params.get(PFPGrowth.MAX_HEAPSIZE)); List<Pair<String,Long>> fList = PFPGrowth.readFList(params); log.info("{}", fList); assertEquals("[(B,6), (D,6), (A,5), (E,4), (C,3)]", fList.toString()); log.info("Starting Grouping Test: {}", params.get(PFPGrowth.MAX_HEAPSIZE)); PFPGrowth.startGroupingItems(params); Map<String,Long> gList = PFPGrowth.deserializeMap(params, PFPGrowth.G_LIST, new Configuration()); log.info("{}", gList); assertEquals("{D=0, E=1, A=0, B=0, C=1}", gList.toString()); log.info("Starting Parallel FPGrowth Test: {}", params.get(PFPGrowth.MAX_HEAPSIZE)); PFPGrowth.startGroupingItems(params); PFPGrowth.startTransactionSorting(params); PFPGrowth.startParallelFPGrowth(params); log.info("Starting Pattern Aggregation Test: {}", params.get(PFPGrowth.MAX_HEAPSIZE)); PFPGrowth.startAggregating(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()); }