How to use MultipleInputFormat in Map Reduce

Hi Friends ,
Today I wanted to give a demo on MultipleInputFormat.In an ideal world we expect our data to come from one source and that too in particular format. for example - if the data exists in  csv file , we expect it to be comma separated and complete data in 1 file . That makes life easy ... isn't it ?

But what if the data exists in multiple different formats and coming from multiple different sources . Lets try to understand this with an example.

Lets say we need to run the following query :

Select name, count(*) from user .
This is a simple query if we need to run this on database but what if the data exists in 2 different formats and 2 different files.
1) CSV Format

Id,Name,Gender
1,Krishna,M
2,Tina,F
3,Umesh,M
4,Nakeeran,F
5,Varun,M
6,Varun,M

2) Tab Format

Id    Name    Gender
7    Tom     M
8    Harry    F
9    Paul    M
10    Nakeeran    F
11    Gopi    M
12    Varun    M

If we want to get the output similar to above written query's output using Mapreduce , we need to look into the different input formats  provided by hadoop.

Lets break the problem in small parts.
1) read the csv file -> we need to create a mapper for this
InputFormat-> TextInputFormat

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

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws java.io.IOException, InterruptedException {
            if (value.toString() != "" && value.toString().contains(",")) {
                String[] array = value.toString().split(",");

                String userInfo = array[0] + "," + array[1] + "," + array[2];

                context.write(new Text(array[1]), new IntWritable(1));

            }
        };
    }

2) read the tab file -> we need to create a mapper for this
InputFormat-> KeyValueTextInputFormat

public static class MiTabMapper extends
            Mapper<Text, Text, Text, IntWritable> {
        @Override
        protected void map(Text key, Text value, Context context)
                throws java.io.IOException, InterruptedException {
            if (value.toString() != "" && value.toString().contains("\t")) {
                String[] array = value.toString().split("\t");
                context.write(new Text(array[0]), new IntWritable(1));
            }
        }
    }

3) Now a simple reducer as if we have only 1 file. Reducer do not care about different sources as long as its getting data in desired format that it can process.

public static class MiReducer extends
            Reducer<Text, IntWritable, Text, IntWritable> {

        @Override
        protected void reduce(Text key, java.lang.Iterable<IntWritable> values,
                Context context) throws java.io.IOException,
                InterruptedException {

            int count = 0;

            for (IntWritable x : values) {
                int val = Integer.parseInt(x.toString());
                count += val;
            }

            context.write(key, new IntWritable(count));

        };
    }

4) But how do we tell Hadoop that we have 2 different formats of input file.
let's take  a look at Driver code.
 MultipleInputs.addInputPath method will take care of multiple files by passing the 4 parameters :
a) job Object
b) Input File path
c) Input    File Format
d) Mapper class

public int run(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Configuration conf = getConf();

        Job job = new Job(conf, "MI Demo");

        job.setJarByClass(MIDemo.class);


        MultipleInputs.addInputPath(job, new Path(args[0]),
                TextInputFormat.class, MiCsvMapper.class);

        MultipleInputs.addInputPath(job, new Path(args[1]),
                KeyValueTextInputFormat.class, MiTabMapper.class);

        job.setReducerClass(MiReducer.class);


        job.setOutputFormatClass(TextOutputFormat.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        // Path in = new Path(args[0]);

        Path out = new Path(args[2]);

        // FileInputFormat.addInputPath(job, in);

        FileOutputFormat.setOutputPath(job, out);

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

        return 0;
    }

5) Now you can run the code with following command. Notice 2 input files and 1 output file.

hadoop jar /home/Desktop/Projects/MultipleInputDemo/MDemo.jar  MIDemo /MultiDemoInput/Users.csv /MultiDemoInput/UsersTab  /MultiDemoOutput/out2

6) Finally the output looks like :

Gopi    1
Harry    1
Krishna    1
Nakeeran    2
Name    2
Paul    1
Tina    1
Tom     1
Umesh    1
Varun    3
 

And that is how you can use MultipleInputFormat . Hope you understood the concept of MultipleInputFormat.

Have a great day...
Happy Hadooping !!!

Varun