Multiple Output Format - Hadoop

Friends,

Hadoop have many output formats  such as :

1) TextOutput Format
2) Sequence File Output Format
3) MapFile Output Format
4) Multiple Output Format

Out of all the available formats, we will talk about Multiple Output formats. I will implement a simple example to demonstrate how it works.

In this example, we will use a sample data which is used for traffic accidents analysis.
We want to calculate following :
1)   Number of accidents per weather condition
2)   Number of accidents per  light condition.

The sample data looks like :

AccidentId, Weather_Condition_Id,Light_Condition_Id
1,3,5,
2,4,3
3,5,6

Explanation of sample data :
Here accident id =1 occurred due to Weather Condition = 3 and Light Condition =5 and so on.

How do we make sure that reducer creates 2 files , one for weather condition and another for Light condition.

I am splitting my csv file on "," and in my scenario 25th column in row is weather condition id and 24th column is Light condition id .

In map method , before writing intermediate key value pairs , I am making sure to attach characters "W" and "L" to distinguish between 2 different columns .

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

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws java.io.IOException, InterruptedException {

            if (value.toString().contains(",")) {
                String[] array = value.toString().split(",");

                context.write(new Text("W" + array[25]), new IntWritable(1));
                context.write(new Text("L" + array[24]), new IntWritable(1));
            }

        };
    }

Now in reducer code, we will create  Multiple Output format in setup method. In reduce method , I used  the character that I attached earlier and based on it writing to 2 different files.

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

        MultipleOutputs<Text, IntWritable> mos;

        @Override
        protected void setup(Context context) throws java.io.IOException,
                InterruptedException {
            mos = new MultipleOutputs<Text, IntWritable>(context);
        };

        @Override
        protected void reduce(Text key, 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;
            }
            if (key.toString().contains("W")) {
                mos.write("Weather",
                        key.toString().substring(1, key.toString().length()),
                        new IntWritable(count));
            } else if (key.toString().contains("L")) {
                mos.write("Light",
                        key.toString().substring(1, key.toString().length()),
                        new IntWritable(count));
            }
        };

        @Override
        protected void cleanup(Context context) throws java.io.IOException,
                InterruptedException {
            mos.close();
        };
    }

Finally, in the driver code, we will configure the naming conventions of output files and output format class for key and value .

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

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

        job.setJarByClass(AnalysisDemo.class);

        job.setMapperClass(AccidentWeatherMapper.class);

        job.setReducerClass(AccidentWeatherReducer.class);


        job.setInputFormatClass(TextInputFormat.class);


        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(IntWritable.class);

//  Here set the name of file and key/value types
        MultipleOutputs.addNamedOutput(job, "Weather", TextOutputFormat.class,
                Text.class, IntWritable.class);

        MultipleOutputs.addNamedOutput(job, "Light", TextOutputFormat.class,
                Text.class, IntWritable.class);

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

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

        FileInputFormat.setInputPaths(job, in);

        FileOutputFormat.setOutputPath(job, out);

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

        return 0;
    }

And that's it. Now you can use a command like this to see data in 2 different files.

hadoop jar /home/Desktop/Projects/Accident/AADemo.jar AnalysisDemo /Accident_Project/Input/Accidents.csv /Accident_Project/Output/Out12

Hope you enjoyed the quick demo of the Multiple Output Format.

Happy Hadooping !!!

Varun