WordCount Example with MapReduce
Published: 2022-07-23 09:15:00

---
title: WordCount Example with MapReduce
published: true
date: 2021-08-14 05:10:29 UTC
tags: 
canonical_url: https://maninekkalapudi.com/wordcount-example-with-mapreduce
---

Hello! Hope you're doing well. In my last [post](https://dev.to/maninekkalapudi/hadoop-mapreduce-a-programming-paradigm-5h4g-temp-slug-1504943) I've explained about internals of Hadoop MapReduce. As promised in that post, we will write and execute a MapReduce program in Java for a simple wordcount example. Let's dive in!

### Topics covered in this post

1. Pre-requisites
2. Hadoop cluster setup on local machine and on Cloud
3. Writing a MapReduce program on Eclipse
4. Create a JAR file for the MapReduce Program and Uploading to HDFS
5. Executing the MapReduce Program on the Hadoop Cluster
6. Results

### 1. Pre-requisites

1. Admin access to the machine (local preferably)
2. Hadoop Cluster (Single/Multi node cluster) on local machine or on cloud
3. Install [JDK 1.8 or later](https://www.oracle.com/in/java/technologies/javase/javase-jdk8-downloads.html) on the local machine
4. [Eclipse IDE](https://www.eclipse.org/downloads/) or any Java IDE installed on the local machine

### 2. Hadoop cluster setup on local machine and on Cloud

**i. Single Node cluster setup**

As we already discussed, the DataNodes store and process the data. We need at least a single node Hadoop cluster to run the MapReduce program and process the data.

Setting up a single node Hadoop cluster on a local machine is a bit lengthy process and often could lead us to errors. I'm sharing the guides that I've used to setup the cluster on my local for testing below for both and Windows and Linux.

- Windows- [https://towardsdatascience.com/installing-hadoop-3-2-1-single-node-cluster-on-windows-10-ac258dd48aef](https://towardsdatascience.com/installing-hadoop-3-2-1-single-node-cluster-on-windows-10-ac258dd48aef)
- Linux (Ubuntu)- [https://phoenixnap.com/kb/install-hadoop-ubuntu](https://phoenixnap.com/kb/install-hadoop-ubuntu)

**ii. Multi node Cluster setup**

Alternatively, we can use a cloud-based Hadoop cluster like [DataProc](https://cloud.google.com/dataproc) on Google Cloud platform (GCP) which doesn't require any setup other than selecting the configuration of the NameNode and the DataNodes. The GCP account setup can be referred [here](https://www.youtube.com/watch?v=W5mPX1-015o). We'll see the setup in the following steps.

Before going any further you should consider two important steps while operating in any cloud environment.

a. Setting up the [billing alerts](https://www.youtube.com/watch?v=F4omjjMZ54k) to avoid any unexpected bills.

b. Turn off/delete the resources soon after the work is done

a. Sign up to the [Google Cloud](https://cloud.google.com/) and login to your account

![Untitled.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628912768681/2_2-FDIWB.png)

b. Search for " **DataProc**" and select the option with the same name in the results

![Untitled 1.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628912893364/S69V9zPhp.png)

c. Select the " **Create Cluster**" option

![Untitled 2.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628912915773/AqE0VVMnE.png)

d. Provide the following details in the create cluster page under "setup a cluster page"

```
i. Cluster name - **test-cluster**

ii. Cluster region and Zone - **us-central1** , **us-central1-a**

iii. Cluster Type - **Standard (1 master, N workers)**

```

![Untitled 3.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913390141/DG7bLloJU.png)

```
iv. Autoscaling Policy - **None**

v. Image type and version - **2.0-debian10 (default)**

vi. **Select Enable Component gateway**

```

![Untitled 4.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913685987/yPy-I9sDR.png)

![Untitled 5.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913703498/C0h2AYaUx.png)

e. Under " **Configure nodes**" select the following for Master node

```
i. Machine family - **General-Purpose (default)**

ii. Series - **N1 (default)**

iii. Machine type - **n1-standard-2 (2 vCPU, 7.5 GB memory)**

iv. Primary disk size (min 15 GB) - **100GB**

v. Primary disk type - **Standard Persistent Disk**

vi. Number of local SSDs - **0**

```

![Untitled 6.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913833256/cKcr_XIwO.png)

f. Select the following for "Worker Nodes"

```
i. Machine family - **General-Purpose (default)**

ii. Series - **N1 (default)**

iii. Machine type - **n1-standard-2 (2 vCPU, 7.5 GB memory)**

iv. Number of worker nodes - **2**

v. Primary disk size (min 15 GB) - **100GB**

vi. Primary disk type - **Standard Persistent Disk**

vii. Number of local SSDs **- 0**

```

![Untitled 7.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913858019/9Y_b3vHrixP.png)

g. Leave the rest of the config as is and select on "CREATE"

![Untitled 8.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913890948/aTYipyWQX.png)

h. Click on the cluster name and select the "VM Instances" tab in the page

![Untitled 9.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913933980/aE70UXDXp.png)

i. Click on "SSH" for the master node and you'll be presented with a new browser window connected to the master node of our HDFS cluster. I've used local terminal to connect to the master node for the rest of the post.

![Untitled 10.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628913980494/qIVrZF_BI.png)

**Note:** In real world scenarios, we would connect to the Hadoop cluster via a gateway node or edge node. We'll not use the NameNode for connecting to the cluster since it'll be very busy in handling the cluster.

### 3. Writing a MapReduce program on Eclipse

a. Create a new Java project called "wordcountmapreduce" in Eclipse IDE on your local machine. Here, I'm using a Linux (ubuntu) machine to create the project. The rest of the steps should stay same for Windows machine as well.

![Untitled 11.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914002133/RA0y2kZlC.png)

b. Create a new Class for Map by right clicking on the project and select "Class". Once you select it, enter the name of the Map class as "WordCountMapper" and hit Finish.

![Untitled 12.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914016333/ZlKA4zogp.png)

c. Once `WordCountMapper` class is created, use the following link for the mapper, reducer, partitioner implementation for the wordcount example. Refer the [GitHub link](https://github.com/maninekkalapudi/dataengineeringbyexamples/blob/main/Hadoop/MapReduce/eclipseprojects/wordcountmapreduce/src/wordcountpackage/WordCountMapper.java) for the code.

![Untitled 13.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914077782/BuL35ZKAT.png)

d. To remove the errors in the IDE, we must mention the Hadoop libraries in project build path. The following are the libraries (only jar files) that should be added to the project:

- `<hadoop_dir>`/share/hadoop/mapreduce (`<hadoop_dir>` is the path where you saved the hadoop distribution. Ex: /home/`<username>`/hadoop-3.3.1)
- `<hadoop_dir>`/share/hadoop/hdfs
- `<hadoop_dir>`/share/hadoop/client
- `<hadoop_dir>`/share/hadoop/common
- `<hadoop_dir>`/share/hadoop/yarn

![Untitled 14.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914131283/3Aq2q4jYgI.png)

Click on "Add External JARs" and navigate to the paths mentioned in the above list. After all the required JARs, click "Apply and Close"

![Untitled 15.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914152194/-sP8awahw.png)

e. After adding the jars to the project build path, we can see the errors disappeared in the IDE in the below image. Use the code for the reducer (WordCountReducer.java), partitioner (WordCountPartitioner.java) and the driver (WordCount.java) classes from the [GitHub link](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/eclipseprojects/wordcountmapreduce/src/wordcountpackage)

![Untitled 16.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914175506/-p_LXluEg.png)

f. Once the project setup is done, we will have a look at the "WordCount.java" class. This is a driver class which executes the Map, Reduce, Combiner and the Partitioner classes on the cluster. This class includes config like

i. Job Name - setJobName ii. Driver class - setJarByClass iii. Mapper class - setMapperClass iv. Combiner class - setCombinerClass. Same as Reducer class for wordcount example v. Reducer class - setReducerClass vi. Number of Reducers - setNumReduceTasks vii. Output data types from each class - setOutputKeyClass, setOutputValueClass viii. Input and Output paths - addInputPath, setOutputPath respectively

![Untitled 17.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914189286/bDF_Ajx4Ys.png)

This is basically the end of the project and code setup required for the wordcount problem in MapReduce.

### 4. Create a JAR file for the MapReduce Program and Uploading to HDFS

Once the project and the Mapreduce code setup is done, there are two ways we could execute the MapReduce Java program:

1. Run the Java program within the eclipse. You can find the guide for the same [here](https://projectgurukul.org/create-hadoop-mapreduce-project-eclipse/). 
2. Package the Java program as a JAR file with all the dependencies and execute on the Hadoop cluster. We'll follow this method for this guide

Steps to package the wordcount MapReduce Java program as a JAR file:

a. Right click on the project and select "Export" option

![Untitled 18.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914624601/YX6KLCX0L.png)

b. Under Java, select "JAR" option and click Next.

![Untitled 19.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914218157/Zq1YKqSVb.png)

c. Select the path for saving the JAR file. Click Next until the final step

![Untitled 20.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914700912/fS9cJp-_N.png)

d. Select the Main class as "WordCount" using Browse window.

![Untitled 21.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914249227/mcdUibZtq.png)

e. Select Finish to create the jar file

![Untitled 22.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914263820/lVsQh02LR.png)

f. The jar file will be created as shown below. Once the jar file is created, we'll upload it to the GCP Hadoop cluster and run it.

![Untitled 23.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628914277414/f66rco09u.png)

g. Now, we'll upload this to the master node in the HDFS cluster using SCP. You can configure SSH to connect to HDFS cluster instance on GCP using this [link](https://www.youtube.com/watch?v=2ibBF9YqveY). I've used Windows + Windows Terminal and the same steps mentioned below are followed. To copy the jar file(s) to master node on the cluster, we use the following command:

```
SCP -i "`<Path/to/SSH/key/ssh-key>`" Path/to/jar/file/wordcountmapperonly.jar username@`<master-ip>`:/path/on/server

```

![Untitled 24.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915047434/nPBEsRKFz.png)

h. Once the jar file is available on the master node instance, we can use the following commands to copy the jar file to the HDFS cluster. Please note master node instance and the HDFS cluster are different.

```
SSH -i "`<Path/to/SSH/key/>`" username@`<master-ip>`
hadoop fs -put -f Path/to/jar/file/wordcountmapperonly.jar `<hdfs_path>`
hadoop fs -ls `<hdfs_path>`

```

![Untitled 25.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915383503/kTFe6Y4UY.png)

Here, we are copying the jar files `wordcountmapperonly.jar`, `wordcountmapreduce.jar` and `wordcountmapreducepartitioner.jar` and the input data folder `HadoopInputFiles` for the Hadoop Directory `'/'`. The input folder contains 3 text files

### 5. Executing the MapReduce Program on the Hadoop Cluster

As we've seen already, the MapReduce driver class (WordCount.java) will be configured to execute Mapper, Combiner, Reducer and Partitioner. We'll run the MapReduce program with different configurations using the driver class

i. Only Mapper

ii. Mapper and Reducer

ii. Mapper, Reducer and Partitioner

**i. Only Mapper**

To run Mapper only, we need to comment out the Combiner, Reducer and Partitioner classes configured in the driver class and package the jar file as shown in the above step. The driver class should look like the below picture. The code for the same is [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/blob/main/Hadoop/MapReduce/eclipseprojects/wordcountmapperonly/src/wordcountpackage/WordCount.java).

![Untitled 26.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915418933/-c9HKnRfM.png)

The input files are in "/HadoopInputFiles" and has data as in three files as mentioned below. You can find the input files [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/HadoopInputFiles).

![Untitled 27.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915438667/hI3ABHCJJ.png)

Now, run the jar file "wordcountmapperonly.jar" on the Hadoop cluster with the following command and above input files. The steps to copy the jar file to HDFS location is shown above section.

```
hadoop jar `<hdfs_path>`/wordcountmapperonly.jar `<input_file_or_dir_path>` `<output_path>`

```

The following image show how to run the mapreduce jars on Hadoop cluster. The full output log of the run is [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/blob/main/Hadoop/MapReduce/logs/mapperonly-log.log).

![Untitled 28.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915462139/ZoGafYArR.png)

The output of the mapper only phase contains all the words with count 1 as shown below

![Untitled 29.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915545193/-GjPz7VsJ.png)

Once we run the MapReduce job, we can see the application is tracked under YARN which is a resource manager for the cluster. Every run gets an entry here. The default YARN URL is **`<cluster-hostname>`:8088**. For DataProc cluster though, we need to go to cluster details in the GCP console, select " **Web Interfaces**" tab under cluster details and select " **YARN ResourceManager**" to get the YARN web interface.

![Untitled 30.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915560469/7f4jqAw71.png)

![Untitled 31.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915573450/4_SWn0iL_.png)

In case where the output path in the `hadoop jar` command already exists, the MapReduce framework throws " **Output directory already exists**" error as shown below. This is to avoid the overwriting of any output data.

![Untitled 32.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915587019/4Q5Pspg0A.png)

**Note:** `-D mapred.reduce.tasks` is set to 3 by default and we need only map phase to run. We can force the reducer count to zero using this property.

In the output path, we can see four different files

1. **\_SUCCESS** - Indicates the job status
2. **part-m-00000 to part-m-00002** - output file corresponding each input files. here 'm' in the output filename indicates 'mapper' phase. Since we don't have a reduce phase configured for this run, we'll get an output file for an input file

![Untitled 33.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915598287/X0LhV63YQ.png)

As we already know, each mapper produces the key-value pairs `<word,1>` for all the words in the input sentence as output shown below

![Untitled 34.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915612631/f-ihJ81jr.png)

**ii. Mapper and Reducer**

Now, Let's run the '[wordcountmapreduce.jar](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/jarfiles)' with the same input files and a different output path. This has both map and reduce phase configured in the driver class. Logs for the run are [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/logs) and code for the same is [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/eclipseprojects/wordcountmapreduce)

![Untitled 35.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915634488/gd7Jl_TIg.png)

The output is generated after the reduce phase into a single output file. Since we have only one reducer by default in the cluster

![Untitled 36.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915647512/d2IJte2vn.png)

![Untitled 37.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915662432/RF4SRauMR.png)

**iii. Mapper, Reducer and Partitioner**

Now, Let's run the '[wordcountmapreducepartitioner.jar](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/jarfiles)' with the same input files and a different output path. This has map, partition and reduce phases configured in the driver class. Logs for the run are [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/logs) and code for the same is [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/tree/main/Hadoop/MapReduce/eclipseprojects/wordcountmapreducepartitioner)

![Untitled 38.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915695861/b6luheBzE.png)

The output for the MapReduce with partitioner is as follows. As per the partitioner logic [here](https://github.com/maninekkalapudi/dataengineeringbyexamples/blob/main/Hadoop/MapReduce/eclipseprojects/wordcountmapreducepartitioner/src/wordcountpackage/WordCountPartitioner.java), for each letter at the starting of the word, there will be a different output file created. This means we are creating 26 partitions which will create same number of reducers to process the records Example: all the words starting with letter 'a' will end up in `'part-r-00001'` file with the count.

![Untitled 39.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915707491/GZ7ZeN-dI.png)

![Untitled 40.png](https://cdn.hashnode.com/res/hashnode/image/upload/v1628915714608/3pywbng6r.png)

### Conclusion

We have seen a practical example of wordcount with MapReduce as promised in my last [post](https://dev.to/maninekkalapudi/hadoop-mapreduce-a-programming-paradigm-5h4g-temp-slug-1504943). This is an exhaustive guide to capture most known ways to create and execute the MapReduce programs in Java.

MapReduce as a compute has lost its edge to new compute framework like Spark. But do you know that we can use the MapReduce to ingest the data into HDFS from an RDBMS source? or write SQL like queries to execute MapReduce job? We will discuss about those in detail in my next blog posts. Stay tuned!

For now though, I'll delete the cloud resource that I've spun up for the tutorial. If you did the same, please delete the resource you have created else you'll end up with something like this

[https://twitter.com/forrestbrazeal/status/1389622850567421952?s=20](https://twitter.com/forrestbrazeal/status/1389622850567421952?s=20)

### Resources

1. [Big Data course](https://trendytech.in/) by [Sumit M](https://www.linkedin.com/in/bigdatabysumit/)
2. https://github.com/maninekkalapudi/dataengineeringbyexamples

If my conent helped you in anyway and like to contribute to my knowledge quest and sharing, you can contribute to me here

Thanks,

Mani