Run Python MapReduce on local Docker Hadoop Cluster
Tags: hadoop, mapreduce
Published: 2020-10-05 21:54:59

## Introduction
This post covers how to deploy a local Docker Hadoop Cluster to run custom Python mapper and reducer function using the classic word count example.

## Environment Setup
**Docker**, get Docker [here](https://docs.docker.com/get-docker/)
**Docker Compose**, get Docker Compose [here](https://docs.docker.com/compose/install/)
**Git**, get Git [here](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)

## Deploy Hadoop Cluster using Docker
We will use the Docker image by [big-data-europe repository](https://github.com/big-data-europe/docker-hadoop) to set up Hadoop.
```
git clone git@github.com:big-data-europe/docker-hadoop.git
```

With the Docker image for Hadoop on your local machine, we can use docker-compose to configure the local Hadoop cluster. Replace the `docker-compose.yml` file with the following file from [this GitHub Gist](https://gist.github.com/nathan815/a938b3f7a4d06b2811cf2b1a917800e1).

This docker-compose file configures a Hadoop cluster with a master node (namenode) and three worker nodes, it also configures the network port to allow communication between the nodes. To start the cluster, run:
```
docker-compose up -d
```
Use `docker ps` to verify the containers are up, you should see a container list similar to the following:
```
IMAGE                           PORTS                    NAMES
docker-hadoop_resourcemanager                            resourcemanager
docker-hadoop_nodemanager1      0.0.0.0:8042->8042/tcp   nodemanager1
docker-hadoop_historyserver     0.0.0.0:8188->8188/tcp   historyserver
docker-hadoop_datanode3         9864/tcp                 datanode3
docker-hadoop_datanode2         9864/tcp                 datanode2
docker-hadoop_datanode1         9864/tcp                 datanode1
docker-hadoop_namenode          0.0.0.0:9870->9870/tcp   namenode
```
The current status of the local Hadoop cluster will be available at [localhost:9870](http://localhost:9870/)


## Running Python MapReduce function
For this simple MapReduce program, we will use the classical word count example. The program reads text files and counts how often each word occurs.

The mapper function will read the text and emit the key-value pair, which in this case is `<word, 1>`. Copy the following code into `mapper.py`
```
#!/usr/bin/env python
"""mapper.py"""

import sys

# input comes from STDIN (standard input)
for line in sys.stdin:
    # remove leading and trailing whitespace
    line = line.strip()
    # split the line into words
    words = line.split()
    # increase counters
    for word in words:
        # write the results to STDOUT (standard output);
        # what we output here will be the input for the
        # Reduce step, i.e. the input for reducer.py
        #
        # tab-delimited; the trivial word count is 1
        print ('%s\t%s' % (word, 1))
```

The reducer function processes the result from the mapper and returns the word count. Copy the following code into `reducer.py`
```
#!/usr/bin/env python
"""reducer.py"""

from operator import itemgetter
import sys

current_word = None
current_count = 0
word = None

# input comes from STDIN
for line in sys.stdin:
    # remove leading and trailing whitespace
    line = line.strip()

    # parse the input we got from mapper.py
    word, count = line.split('\t', 1)

    # convert count (currently a string) to int
    try:
        count = int(count)
    except ValueError:
        # count was not a number, so silently
        # ignore/discard this line
        continue

    # this IF-switch only works because Hadoop sorts map output
    # by key (here: word) before it is passed to the reducer
    if current_word == word:
        current_count += count
    else:
        if current_word:
            # write result to STDOUT
            print ('%s\t%s' % (current_word, current_count))
        current_count = count
        current_word = word

# do not forget to output the last word if needed!
if current_word == word:
    print ('%s\t%s' % (current_word, current_count))
```

Note because Hadoop runs on Apache server which is built in Java, the program takes a Java JAR file as an input. To execute Python in Hadoop, we will need to use the [Hadoop Streaming library ](https://hadoop.apache.org/docs/r1.2.1/streaming.html) to pipe the Python executable into the Java framework. As a result, we need to process the Python input from STDIN.

Copy the local `mapper.py` and `reducer.py` to the namenode:
```
docker cp LOCAL_PATH/mapper.py namenode:mapper.py
docker cp LOCAL_PATH/reducer.py namenode:reducer.py
```

Enter the namenode container of the Hadoop cluster:
```
docker exec -it namenode bash
```

Run `ls` and you should find `mapper.py` and `reducer.py` in the namenode container.

Now let's prepare the input. For this simple example, we will use a set of text files with a short string. For a more realistic example, you can use e-book from [Project Gutenberg](http://www.gutenberg.org/), download the `Plain Text UTF-8` encoding.
```
mkdir input
echo "Hello World" >input/f1.txt
echo "Hello Docker" >input/f2.txt
echo "Hello Hadoop" >input/f3.txt
echo "Hello MapReduce" >input/f4.txt
```
The MapReduce program access files from the Hadoop Distributed File System (HDFS). Run the following to transfer the input directory and files to HDFS:
```
hadoop fs -mkdir -p input
hdfs dfs -put ./input/* input
```

Use `find / -name 'hadoop-streaming*.jar'` to locate the hadoop string library JAR file. The path should look something like `PATH/hadoop-streaming-3.2.1.jar`
Finally, we can execute the MapReduce program:

```
hadoop jar /opt/hadoop-3.2.1/share/hadoop/tools/lib/hadoop-streaming-3.2.1.jar \
-file mapper.py    -mapper mapper.py \
-file reducer.py   -reducer reducer.py \
-input input -output output
```


To safely shut down the cluster and remove containers, run: ```
docker-compose down
```


## Reference
Yen V. (2019). How to set up a Hadoop cluster in Docker.
Retrieved from: [here](![Screen Shot 2020-10-05 at 5.47.33 PM](https://dev-to-uploads.s3.amazonaws.com/i/2sd0llh04f4lmn0imryw.png)
 )

Noll M. Writing An Hadoop MapReduce Program In Python.
Retrieved from: [here](https://www.michael-noll.com/tutorials/writing-an-hadoop-mapreduce-program-in-python/)