package com.macksweeney;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * Created by msweeney on 9/10/16.
 */
public class PatentCiteCountHistogram extends Configured implements Tool {

    private static class PatentCitingCounterMapper extends Mapper<Text, Text, Text, Text> {

        /** Reverse citing -> cited relationship to cited -> citing. */
        @Override
        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
            context.write(value, key);
        }
    }

    private static class PatentCitingCounterReducer extends Reducer<Text, Text, Text, IntWritable> {

        /** For each patent, output the number of patents citing it. */
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException
        {
            int count = 0;
            for (Text citing : values) {
                count += 1;
            }
            context.write(key, new IntWritable(count));
        }
    }

    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");
        Job job = Job.getInstance(conf, PatentCiteCountHistogram.class.getSimpleName());
        job.setJarByClass(PatentCiteCountHistogram.class);

        Path in = new Path(args[0]);
        Path out = new Path(args[1]);
        FileInputFormat.setInputPaths(job, in);
        FileOutputFormat.setOutputPath(job, out);

        job.setMapperClass(PatentCitingCounterMapper.class);
        job.setReducerClass(PatentCitingCounterReducer.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new Configuration(), new PatentCiteCountHistogram(), args);
        System.exit(exitCode);
    }
}

@@ -859,34 +859,39 @@
 ss PatentCit
-ing
+e
 Count
-er
+Histogram
 Mapper exten
@@ -916,90 +916,244 @@
 xt, 
-Text, Text%3E %7B%0A%0A        /** Reverse citing -%3E cited relationship to cited -%3E citing
+IntWritable, IntWritable%3E %7B%0A%0A        private final static IntWritable one = new IntWritable(1);%0A        private IntWritable citationCount = new IntWritable();%0A%0A        /** Output citation count and 1; these will be summed in the reducer
 . */
@@ -1297,31 +1297,106 @@
    c
-ontext.write(value, key
+itationCount.set(Integer.parseInt(value.toString()));%0A            context.write(citationCount, one
 );%0A 
@@ -1441,34 +1441,39 @@
 ss PatentCit
-ing
+e
 Count
-er
+Histogram
 Reducer exte
@@ -1479,118 +1479,220 @@
 ends
- Reducer%3CText, Text, Text, IntWritable%3E %7B%0A%0A        /** For each patent, output the number of patents citing it
+%0A            Reducer%3CIntWritable, IntWritable, IntWritable, IntWritable%3E %7B%0A%0A        private IntWritable countOfCounts = new IntWritable();%0A%0A        /** Sum how many times each count showed up to get the histogram
 . */
@@ -1740,20 +1740,27 @@
  reduce(
-Text
+IntWritable
  key, It
@@ -1766,20 +1766,27 @@
 terable%3C
-Text
+IntWritable
 %3E values
@@ -1919,19 +1919,25 @@
 or (
-Text citing
+IntWritable value
  : v
@@ -1970,17 +1970,27 @@
 ount += 
-1
+value.get()
 ;%0A      
@@ -2015,47 +2015,76 @@
   co
-ntext.write(key, new IntWritable(c
+untOfCounts.set(count);%0A            context.write(key, countOfC
 ount
-)
+s
 );%0A 
@@ -2091,32 +2091,134 @@
        %7D%0A    %7D%0A%0A
+    /** Default key, value separator is the tab, which is appropriate for the citation count data. */%0A
     public int r
@@ -2298,95 +2298,8 @@
 ();%0A
-        conf.set(%22mapreduce.input.keyvaluelinerecordreader.key.value.separator%22, %22,%22);%0A
     
@@ -2653,26 +2653,31 @@
 atentCit
-ing
+e
 Count
-er
+Histogram
 Mapper.c
@@ -2724,18 +2724,23 @@
 tCit
-ing
+e
 Count
-er
+Histogram
 Redu
@@ -2900,28 +2900,35 @@
 putKeyClass(
-Text
+IntWritable
 .class);%0A   
@@ -2956,20 +2956,27 @@
 ueClass(
-Text
+IntWritable
 .class);

add class to build histogram of patent citation counts