Hadoop MapReduce Join & Counter with Example

By : Prafulla Ranadive Prafulla Ranadive
Hours Updated

⚡ Smart Summary

MapReduce joins combine two large datasets on a shared key, either inside the mapper or inside the reducer, while MapReduce counters collect statistics about the job so that bad records can be measured rather than guessed.

  • 🔘 Join basics: The smaller of the two datasets is distributed to every data node and used as the lookup side.
  • ☑️ Map-side join: Requires each input to be partitioned, equally split and sorted by the join key before the map function runs.
  • ✅ Reduce-side join: Needs no partitioning, because every tuple sharing a join key lands in the same reducer.
  • 🧪 Worked example: DeptName.txt and DeptStrength.txt are copied into HDFS and joined on Dept_ID by a packaged jar.
  • 🛠️ Counter types: Five built-in counter groups ship with every job, and user-defined counters are declared as a Java enum.
  • ⚠️ Counter use: Incrementing a counter on each missing or invalid record turns data-quality problems into a number on the job report.
Read More

Hadoop MapReduce join and counter tutorial with a worked example

What is Join in MapReduce?

MapReduce Join operation is used to combine two large datasets. However, this process involves writing lots of code to perform the actual join operation. Joining two datasets begins by comparing the size of each dataset. If one dataset is smaller as compared to the other dataset, then the smaller dataset is distributed to every data node in the cluster.

Once a join in MapReduce is distributed, either the Mapper or the Reducer uses the smaller dataset to perform a lookup for matching records from the large dataset and then combines those records to form output records.

Types of Join

Depending upon the place where the actual join is performed, joins in Hadoop are classified into two kinds.

  1. Map-side join — When the join is performed by the mapper, it is called a map-side join. In this type, the join is performed before data is actually consumed by the map function. It is mandatory that the input to each map is in the form of a partition and is in sorted order. Also, there must be an equal number of partitions and it must be sorted by the join key.
  2. Reduce-side join — When the join is performed by the reducer, it is called a reduce-side join. There is no necessity in this join to have a dataset in a structured form (or partitioned). Here, map side processing emits the join key and the corresponding tuples of both the tables. As an effect of this processing, all the tuples with the same join key fall into the same reducer, which then joins the records with the same join key.

An overall process flow of joins in Hadoop is depicted in below diagram.

Process flow diagram comparing a map-side join with a reduce-side join in Hadoop
Types of Joins in Hadoop MapReduce

With the two variants clear, the next section walks through a reduce-side join on two small department files.

How to Join two DataSets: MapReduce Example

There are two Sets of Data in two Different Files (shown below). The Key Dept_ID is common in both files. The goal is to use MapReduce Join to combine these files.

First input file listing department IDs alongside department names

File 1
Second input file listing department IDs alongside department strength values

File 2

Input: The input data set is a txt file, DeptName.txt & DeptStrength.txt

Download Input Files From Here

Ensure you have Hadoop installed. Before you start with the MapReduce Join example actual process, change user to ‘hduser’ (id used while Hadoop configuration, you can switch to the userid used during your Hadoop config).

su - hduser_

The prompt changes to the Hadoop account, as shown below.

Terminal after switching to the hduser account with the su command

Step 1) Copy the zip file to the location of your choice

Downloaded MapReduceJoin archive placed in the chosen working directory

Step 2) Uncompress the Zip File

sudo tar -xvf MapReduceJoin.tar.gz

The extracted file names scroll past as tar unpacks the archive.

Console listing of files extracted from MapReduceJoin.tar.gz

Step 3) Go to directory MapReduceJoin/

cd MapReduceJoin/

Shell prompt after changing into the MapReduceJoin directory

Step 4) Start Hadoop

$HADOOP_HOME/sbin/start-dfs.sh
$HADOOP_HOME/sbin/start-yarn.sh

Both scripts print the daemons they bring up.

Startup messages from the HDFS and YARN daemon scripts

Step 5) DeptStrength.txt and DeptName.txt are the input files used for this MapReduce Join example program.

These files need to be copied to HDFS using below command-

$HADOOP_HOME/bin/hdfs dfs -copyFromLocal DeptStrength.txt DeptName.txt /

Both input text files copied into the HDFS root directory

RELATED ARTICLES

Step 6) Run the program using below command-

$HADOOP_HOME/bin/hadoop jar MapReduceJoin.jar MapReduceJoin/JoinDriver/DeptStrength.txt /DeptName.txt /output_mapreducejoin

The command is echoed first, and the job then reports its progress on the console.

Command line launching the packaged MapReduceJoin jar file

Console output tracking the progress of the MapReduce join job

Step 7) After execution, the output file (named ‘part-00000’) will be stored in the directory /output_mapreducejoin on HDFS

Results can be seen using the command line interface

$HADOOP_HOME/bin/hdfs dfs -cat /output_mapreducejoin/part-00000

Joined department records printed from HDFS by the cat command

Results can also be seen via a web interface as-

Hadoop web interface landing page used to reach the file system browser

Now select ‘Browse the filesystem’ and navigate up to /output_mapreducejoin

Browsing the HDFS file system view to the output_mapreducejoin directory

Open part-r-00000

Selecting the part-r-00000 output file inside the browser view

Results are shown

Joined department name and department strength rows displayed in the browser

NOTE: Please note that before running this program for the next time, you will need to delete output directory /output_mapreducejoin

$HADOOP_HOME/bin/hdfs dfs -rm -r /output_mapreducejoin

Alternative is to use a different name for the output directory.

Joins tell you what the data looks like. Counters, covered next, tell you how the job that produced it behaved.

What is Counter in MapReduce?

A Counter in MapReduce is a mechanism used for collecting and measuring statistical information about MapReduce jobs and events. Counters keep the track of various job statistics in MapReduce like number of operations occurred and progress of the operation. Counters are used for problem diagnosis in MapReduce.

Hadoop Counters are similar to putting a log message in the code for a map or reduce. This information could be useful for diagnosis of a problem in MapReduce job processing.

Typically, these counters in Hadoop are defined in a program (map or reduce) and are incremented during execution when a particular event or condition (specific to that counter) occurs. A very good application of Hadoop counters is to track valid and invalid records from an input dataset.

Types of MapReduce Counters

There are basically 2 types of MapReduce Counters

  1. Hadoop built-in counters: There are some built-in Hadoop counters which exist per job. Below are built-in counter groups-
    • MapReduce Task Counters — Collects task specific information (e.g., number of input records) during its execution time.
    • FileSystem Counters — Collects information like number of bytes read or written by a task.
    • FileInputFormat Counters — Collects information of a number of bytes read through FileInputFormat.
    • FileOutputFormat Counters — Collects information of a number of bytes written through FileOutputFormat.
    • Job Counters — These counters record job-wide statistics such as the number of tasks launched for a job.
  2. User defined counters: In addition to built-in counters, a user can define his own counters using similar functionalities provided by programming languages. For example, in Java, an ‘enum’ is used to define user defined counters.

💡 Version note: job counters were maintained by the JobTracker under MRv1. On YARN that role belongs to the MapReduce ApplicationMaster, so the counter names survive but the component reporting them has changed.

A job cannot declare an unlimited number of counters. The mapreduce.job.counters.max setting caps the total per job at 120 by default, and a job that declares more fails with a LimitExceededException, so counters are meant for a handful of aggregate signals rather than per-key tallies.

Counters Example

An example MapClass with Counters to count the number of missing and invalid values. Input data file used in this tutorial Our input data set is a CSV file, SalesJan2009.csv

public static class MapClass
            extends MapReduceBase
            implements Mapper<LongWritable, Text, Text, Text>
{
    static enum SalesCounters { MISSING, INVALID };
    public void map ( LongWritable key, Text value,
                 OutputCollector<Text, Text> output,
                 Reporter reporter) throws IOException
    {
        
        //Input string is split using ',' and stored in 'fields' array
        String fields[] = value.toString().split(",", -20);
        //Value at 4th index is country. It is stored in 'country' variable
        String country = fields[4];
        
        //Value at 8th index is sales data. It is stored in 'sales' variable
        String sales = fields[8];
      
        if (country.length() == 0) {
            reporter.incrCounter(SalesCounters.MISSING, 1);
        } else if (sales.startsWith("\"")) {
            reporter.incrCounter(SalesCounters.INVALID, 1);
        } else {
            output.collect(new Text(country), new Text(sales + ",1"));
        }
    }
}

Above code snippet shows an example implementation of counters in Hadoop MapReduce.

Here, SalesCounters is a counter defined using ‘enum’. It is used to count MISSING and INVALID input records.

In the code snippet, if ‘country’ field has zero length then its value is missing and hence corresponding counter SalesCounters.MISSING is incremented.

Next, if ‘sales’ field starts with a “ then the record is considered INVALID. This is indicated by incrementing counter SalesCounters.INVALID.

💡 API note: the snippet above uses the original org.apache.hadoop.mapred API, where MapReduceBase, the Mapper interface, OutputCollector and Reporter appear separately. Current code is written against org.apache.hadoop.mapreduce, where a single Context replaces the collector and the reporter, and a counter is incremented with context.getCounter(SalesCounters.MISSING).increment(1). The counter concept is identical in both.

FAQs

Pick the mapper variant when one side is small enough to hold in memory on every node, because it skips the shuffle entirely. Pick the reducer variant when both sides are large or unsorted, and accept the extra network cost.

Models learn from historical job history to predict runtime, recommend split sizes and reducer counts, and detect skew from counter values. They also flag jobs whose spilled-record or failed-task counters drift outside the normal band for that pipeline.

Copilot produces plausible mapper and reducer skeletons, but it freely mixes the old mapred package with the newer mapreduce package in one class, which will not compile. Fix the imports and the method signatures before trusting the logic.

It is the mechanism that ships the smaller file to every node before the tasks start. Each mapper then loads that copy into a hash map and looks up matches locally, which is what makes a mapper-side join possible.

They are printed in the console summary when the job completes, exposed in the job history and resource manager web pages, and readable programmatically from the job object, so a driver can assert on them and fail a bad run.

A single Context object. It carries the work that OutputCollector and Reporter used to split between them, so output is written and counters are incremented through the same handle passed into the map method.

For most reporting work, no. A HiveQL join compiles down to the same shuffle-and-merge pattern in a few lines. Hand-written jobs are worth it when the merge logic does not fit a SQL clause.

Hadoop refuses to write into an output directory that already exists, which protects finished results from being overwritten. Delete the directory recursively first, or pass a different output path on the next run.

Summarize this post with: