@@ -0,0 +1,3025 @@
+package com.isuraed.insight;%0A%0Aimport java.io.IOException;%0A%0Aimport org.apache.hadoop.conf.Configuration;%0Aimport org.apache.hadoop.fs.Path;%0Aimport org.apache.hadoop.io.LongWritable;%0Aimport org.apache.hadoop.io.Text;%0Aimport org.apache.hadoop.mapreduce.Job;%0Aimport org.apache.hadoop.mapreduce.lib.input.FileInputFormat;%0Aimport org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;%0Aimport org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;%0Aimport org.apache.hadoop.mapreduce.Mapper;%0Aimport org.apache.hadoop.mapreduce.Reducer;%0A%0Aimport org.apache.log4j.Logger;%0A%0Apublic class ReviewsByProductTitle %7B%0A%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        Configuration conf = new Configuration();%0A        Job job = new Job(conf, %22isura_ReviewsByProductTitle%22);%0A%0A        job.setJarByClass(ReviewsByProductTitle.class);%0A        job.setMapperClass(ReviewsMapper.class);%0A        job.setReducerClass(ReviewsReducer.class);%0A%0A        job.setOutputKeyClass(Text.class);%0A        job.setOutputValueClass(Text.class);%0A        job.setMapOutputKeyClass(Text.class);%0A        job.setMapOutputValueClass(Text.class);%0A%0A        job.setOutputFormatClass(TextOutputFormat.class);%0A        FileInputFormat.addInputPath(job, new Path(args%5B0%5D));%0A        FileOutputFormat.setOutputPath(job, new Path(args%5B1%5D));%0A%0A        boolean result = job.waitForCompletion(true);%0A        System.exit(result ? 0 : 1);%0A    %7D%0A%0A    // Earlier in the pipeline we already calculated the title for each productId. The mapper simply reverses the key and value.%0A    static class ReviewsMapper extends Mapper%3CLongWritable,Text,Text,Text%3E %7B%0A        private static final Logger logger = Logger.getLogger(ReviewsMapper.class.getName());%0A%0A        @Override%0A        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException %7B%0A            String%5B%5D row = value.toString().split(%22%5Ct%22, -1);%0A            String productId = row%5B0%5D.toString();%0A            String title = row%5B1%5D.toString();%0A%0A            if (productId.equals(%22%22)) %7B%0A                logger.warn(%22Skipped record: productId was blank%22);%0A                return;%0A            %7D%0A%0A            if (title.equals(%22%22)) %7B%0A                logger.warn(%22Skipped record: Title was blank%22);%0A                return;%0A            %7D%0A%0A            // Escape quotes because strings are stored in json later.%0A            title = title.replace(%22%5C%22%22, %22%5C%5C%5C%22%22);%0A            title = title.toLowerCase();%0A%0A            context.write(new Text(title), new Text(productId));%0A        %7D%0A    %7D%0A%0A    static class ReviewsReducer extends Reducer%3CText, Text, Text, Text%3E %7B%0A%0A        @Override%0A        protected void reduce(Text key, Iterable%3CText%3E values, Context context) throws IOException, InterruptedException %7B%0A            StringBuilder productIds = new StringBuilder();%0A%0A            for (Text val : values) %7B%0A                productIds.append(val);%0A                productIds.append(%22%5Ct%22);%0A            %7D%0A%0A            context.write(key, new Text(productIds.toString().trim()));%0A        %7D%0A    %7D%0A%7D%0A

Build reverse map from title to productId