Multiple Input Files Mapreduce Wordcount example done separately

Each Reducer outputs one output file. The number of output files is dependent on number of Reducers.

(A) Assuming you want to process all three input files in a single MapReduce Job.

At the very minimum - you must set number of Reducers equal to the Number of Output Files you want.

Since you are trying to do word-counts Per File. And not across Files. You will have to ensure that all the file contents (of one file) are processed by a Single Reducer. Using a Custom Partitioner is one way to do this.

(B) Another way is to simply run your MapReduce Job Three Times. Once for Each Input File. And have Reducer count as 1.

Even I am a newbie in hadoop and found this question very interesting. And this is how I resolved this.

public class Multiwordcnt {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException  {

            Configuration conf = new Configuration();
            Job myJob = new Job(conf, "Multiwordcnt");
            String[] userargs = new GenericOptionsParser(conf, args).getRemainingArgs();

            myJob.setJarByClass(Multiwordcnt.class);
            myJob.setMapperClass(MyMapper.class);
            myJob.setReducerClass(MyReducer.class);     
            myJob.setMapOutputKeyClass(Text.class);
            myJob.setMapOutputValueClass(IntWritable.class);

            myJob.setOutputKeyClass(Text.class);
            myJob.setOutputValueClass(IntWritable.class);

            myJob.setInputFormatClass(TextInputFormat.class);
            myJob.setOutputFormatClass(TextOutputFormat.class);

            FileInputFormat.addInputPath(myJob, new Path(userargs[0]));
            FileOutputFormat.setOutputPath(myJob, new Path(userargs[1]));

            System.exit(myJob.waitForCompletion(true) ? 0 : 1 );
    }

    public static  class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        Text emitkey = new Text();
        IntWritable emitvalue = new IntWritable(1);

        public void map(LongWritable key , Text value, Context context) throws IOException, InterruptedException {

            String filePathString = ((FileSplit) context.getInputSplit()).getPath().toString();                     
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while  (tokenizer.hasMoreTokens()){

                String filepathword = filePathString + "*" + tokenizer.nextToken();
                emitkey.set(filepathword);
                context.write(emitkey, emitvalue);
            }           
        }
    }

    public static  class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        Text emitkey = new Text();
        IntWritable emitvalue = new IntWritable();
        private MultipleOutputs<Text,IntWritable> multipleoutputs;

        public void setup(Context context) throws IOException, InterruptedException {
            multipleoutputs = new MultipleOutputs<Text,IntWritable>(context);
        }           

        public void reduce(Text key , Iterable <IntWritable> values, Context context)   throws IOException, InterruptedException {
            int sum = 0;

            for (IntWritable value : values){
                sum = sum + value.get();
            }
            String pathandword = key.toString();
            String[] splitted = pathandword.split("\\*");
            String path = splitted[0];
            String word = splitted[1];              
            emitkey.set(word);
            emitvalue.set(sum);
            System.out.println("word:" + word + "\t" + "sum:" + sum + "\t" + "path:  " + path);
            multipleoutputs.write(emitkey,emitvalue , path);
        }

        public void cleanup(Context context) throws IOException, InterruptedException {
            multipleoutputs.close();
        }
    }
}