{"id":109209,"date":"2021-02-18T17:12:24","date_gmt":"2021-02-18T11:42:24","guid":{"rendered":"https:\/\/www.vskills.in\/certification\/tutorial\/?page_id=109209"},"modified":"2024-04-12T14:30:28","modified_gmt":"2024-04-12T09:00:28","slug":"combiner-functions","status":"publish","type":"page","link":"https:\/\/www.vskills.in\/certification\/tutorial\/combiner-functions\/","title":{"rendered":"Combiner Functions"},"content":{"rendered":"\n

A Combiner, also known as a semi-reducer, is an optional class that operates by accepting the inputs from the Map class and thereafter passing the output key-value pairs to the Reducer class. The main function of a Combiner is to summarize the map output records with the same key. The output (key-value collection) of the combiner will be sent over the network to the actual Reducer task as input.<\/p>\n\n\n\n

The Combiner class is used in between the Map class and the Reduce class to reduce the volume of data transfer between Map and Reduce. Usually, the output of the map task is large and the data transferred to the reduce task is high. The following MapReduce task diagram shows the COMBINER PHASE.<\/p>\n\n\n\n

\"combiner-functions\"<\/figure><\/div>\n\n\n\n

A combiner does not have a predefined interface and it must implement the Reducer interface\u2019s reduce() method. A combiner operates on each map output key. It must have the same output key-value types as the Reducer class. A combiner can produce summary information from a large dataset because it replaces the original Map output. Although, Combiner is optional yet it helps segregating data into multiple groups for Reduce phase, which makes it easier to process.<\/p>\n\n\n\n

Combiner Advantage<\/strong><\/p>\n\n\n\n

When a MapReduce Job is run on a large dataset, Hadoop Mapper generates large chunks of intermediate data that is passed on to Hadoop Reducer for further processing, which leads to massive network congestion. So how do go about reducing this network congestion? Is there any function in Hadoop to address this issue? The MapReduce framework offers a function known as \u2018Combiner\u2019 that can play a crucial role in reducing network congestion. As a matter of fact \u2018Combiner\u2019 is also termed as \u2018Mini-reducer\u2019. It is important to note that the primary job of a Hadoop Combiner is to process the output data from Hadoop Mapper, before passing it to a Hadoop Reducer. Technically speaking, Combiner and Reducer use the same code.<\/p>\n\n\n\n

In order to understand the concept of Hadoop Combiner effectively, let\u2019s consider the following use case, which identifies the number of complaints reported in each state. Below is the code snippet<\/p>\n\n\n\n

package com.evoke.bigdata.mr.complaint;<\/p>\n\n\n\n

import org.apache.hadoop.fs.Path;<\/p>\n\n\n\n

import org.apache.hadoop.io.IntWritable;<\/p>\n\n\n\n

import org.apache.hadoop.io.Text;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.FileInputFormat;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.FileOutputFormat;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.JobClient;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.JobConf;<\/p>\n\n\n\n

import org.apache.hadoop.conf.Configured;<\/p>\n\n\n\n

import org.apache.hadoop.util.Tool;<\/p>\n\n\n\n

import org.apache.hadoop.util.ToolRunner;<\/p>\n\n\n\n

public class ComplaintCombiner extends Configured implements Tool {<\/p>\n\n\n\n

@Override<\/p>\n\n\n\n

public int run(String[] args) throws Exception {<\/p>\n\n\n\n

String input, output;<\/p>\n\n\n\n

if(args.length == 2) {<\/p>\n\n\n\n

input = args[0];<\/p>\n\n\n\n

output = args[1];<\/p>\n\n\n\n

} else {<\/p>\n\n\n\n

input = “your-input-dir”;<\/p>\n\n\n\n

output = “your-output-dir”;<\/p>\n\n\n\n

}<\/p>\n\n\n\n

JobConf conf = new JobConf(getConf(), ComplaintCombiner.class);<\/p>\n\n\n\n

conf.setJobName(this.getClass().getName());<\/p>\n\n\n\n

FileInputFormat.setInputPaths(conf, new Path(input));<\/p>\n\n\n\n

FileOutputFormat.setOutputPath(conf, new Path(output));<\/p>\n\n\n\n

conf.setMapperClass(StateMapper.class);<\/p>\n\n\n\n

conf.setReducerClass(SumCombiner.class);<\/p>\n\n\n\n

\/\/conf.setCombinerClass(SumCombiner.class);<\/p>\n\n\n\n

conf.setMapOutputKeyClass(Text.class);<\/p>\n\n\n\n

conf.setMapOutputValueClass(IntWritable.class);<\/p>\n\n\n\n

conf.setOutputKeyClass(Text.class);<\/p>\n\n\n\n

conf.setOutputValueClass(IntWritable.class);<\/p>\n\n\n\n

JobClient.runJob(conf);<\/p>\n\n\n\n

return 0;<\/p>\n\n\n\n

}<\/p>\n\n\n\n

public static void main(String[] args) throws Exception {<\/p>\n\n\n\n

int exitCode = ToolRunner.run(new ComplaintCombiner(), args);<\/p>\n\n\n\n

System.exit(exitCode);<\/p>\n\n\n\n

}<\/p>\n\n\n\n

}<\/p>\n\n\n\n

Below is the mapper code that can be used to retrieve complaints for each state:<\/p>\n\n\n\n

package com.evoke.bigdata.mr.complaint;<\/p>\n\n\n\n

import java.io.IOException;<\/p>\n\n\n\n

import org.apache.hadoop.io.IntWritable;<\/p>\n\n\n\n

import org.apache.hadoop.io.LongWritable;<\/p>\n\n\n\n

import org.apache.hadoop.io.Text;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.MapReduceBase;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.Mapper;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.OutputCollector;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.Reporter;<\/p>\n\n\n\n

public class StateMapper extends MapReduceBase implements<\/p>\n\n\n\n

Mapper<LongWritable, Text, Text, IntWritable> {<\/p>\n\n\n\n

@Override<\/p>\n\n\n\n

public void map(LongWritable key, Text value,<\/p>\n\n\n\n

OutputCollector<Text, IntWritable> output, Reporter reporter)<\/p>\n\n\n\n

throws IOException {<\/p>\n\n\n\n

String s = value.toString();<\/p>\n\n\n\n

String[] fields = value.toString().split(“,”);<\/p>\n\n\n\n

if (fields.length > 1) {<\/p>\n\n\n\n

output.collect(new Text(fields[5]), new IntWritable(1));<\/p>\n\n\n\n

}<\/p>\n\n\n\n

}<\/p>\n\n\n\n

}<\/p>\n\n\n\n

And lastly, here\u2019s the Reducer\/Combiner code that provides the total number of complaints received in each state:<\/p>\n\n\n\n

package com.evoke.bigdata.mr.complaint;<\/p>\n\n\n\n

import java.io.IOException;<\/p>\n\n\n\n

import java.util.Iterator;<\/p>\n\n\n\n

import org.apache.hadoop.io.IntWritable;<\/p>\n\n\n\n

import org.apache.hadoop.io.Text;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.OutputCollector;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.MapReduceBase;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.Reducer;<\/p>\n\n\n\n

import org.apache.hadoop.mapred.Reporter;<\/p>\n\n\n\n

public class SumCombiner extends MapReduceBase implements<\/p>\n\n\n\n

Reducer<Text, IntWritable, Text, IntWritable> {<\/p>\n\n\n\n

@Override<\/p>\n\n\n\n

public void reduce(Text key, Iterator<IntWritable> values,<\/p>\n\n\n\n

OutputCollector<Text, IntWritable> output, Reporter reporter)<\/p>\n\n\n\n

throws IOException {<\/p>\n\n\n\n

int stateCount = 0;<\/p>\n\n\n\n

while (values.hasNext()) {<\/p>\n\n\n\n

IntWritable value = values.next();<\/p>\n\n\n\n

stateCount += value.get();<\/p>\n\n\n\n

}<\/p>\n\n\n\n

output.collect(key, new IntWritable(stateCount));<\/p>\n\n\n\n

}<\/p>\n\n\n\n

}<\/p>\n\n\n\n

In like manner, if the same job is executed using Hadoop Combiner by replacing setReducerClass method with a setCombinerClass method in the job code, it will drastically reduce the number of bytes transmitted over the network. Furthermore, we can clearly observe in the Job Tracker UI that very limited data is transferred to the Reduce phase<\/p>\n\n\n\n

Combiner Example<\/strong><\/p>\n\n\n\n

The following example provides a theoretical idea about combiners. Let us assume we have the following input text file named input.txt for MapReduce.<\/p>\n\n\n\n

What do you mean by Object<\/p>\n\n\n\n

What do you know about Java<\/p>\n\n\n\n

What is Java Virtual Machine<\/p>\n\n\n\n

How Java enabled High Performance<\/p>\n\n\n\n

The important phases of the MapReduce program with Combiner are<\/p>\n\n\n\n

Record Reader<\/strong> – This is the first phase of MapReduce where the Record Reader reads every line from the input text file as text and yields output as key-value pairs.<\/p>\n\n\n\n