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
Docker Compose, get Docker Compose here
Git, get Git here
Deploy Hadoop Cluster using Docker
We will use the Docker image by big-data-europe repository 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.
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
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 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, 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
Noll M. Writing An Hadoop MapReduce Program In Python.
Retrieved from: here
){kind=link}
Top comments (1)
Thank you for a great tutorial. I tried to run like your tutorial and everything work as expected but I cannot run the last MapReduce program. Is it because I have not installed python to the
namenoderight?And it returns the error so where can I find the
stderrfile to know what is the error with my code?