Home > Hadoop, MapReduce, Python > Hadoop Streaming Made Simple using Joins and Keys with Python

Hadoop Streaming Made Simple using Joins and Keys with Python

There are a lot of different ways to write MapReduce jobs!!!

Sample code for this post https://github.com/joestein/amaunet

I find streaming scripts a good way to interrogate data sets (especially when I have not worked with them yet or are creating new ones) and enjoy the lifecycle when the initial elaboration of the data sets lead to the construction of the finalized scripts for an entire job (or series of jobs as is often the case).

When doing streaming with Hadoop you do have a few library options.  If you are a Ruby programmer then wukong is awesome! For Python programmers you can use dumbo and more recently released mrjob.

I like working under the hood myself and getting down and dirty with the data and here is how you can too.

Lets start first with defining two simple sample data sets.

Data set 1:  countries.dat


United States|US
United Kingdom|UK

Data set 2: customers.dat


Alice Bob|not bad|US
Sam Sneed|valued|CA
Jon Sneed|valued|CA
Arnold Wesise|not so good|UK
Henry Bob|not bad|US
Yo Yo Ma|not so good|CA
Jon York|valued|CA
Alex Ball|valued|UK
Jim Davis|not so bad|JA

The requirements: you need to find out grouped by type of customer how many of each type are in each country with the name of the country listed in the countries.dat in the final result (and not the 2 digit country name).

To-do this you need to:

1) Join the data sets
2) Key on country
3) Count type of customer per country
4) Output the results

So first lets code up a quick mapper called smplMapper.py (you can decide if smpl is short for simple or sample).

Now in coding the mapper and reducer in Python the basics are explained nicely here http://www.michael-noll.com/tutorials/writing-an-hadoop-mapreduce-program-in-python/ but I am going to dive a bit deeper to tackle our example with some more tactics.

#!/usr/bin/env python

import sys

# input comes from STDIN (standard input)
for line in sys.stdin:
	try: #sometimes bad data can cause errors use this how you like to deal with lint and bad data
		personName = "-1" #default sorted as first
		personType = "-1" #default sorted as first
		countryName = "-1" #default sorted as first
		country2digit = "-1" #default sorted as first
		# remove leading and trailing whitespace
		line = line.strip()
		splits = line.split("|")
		if len(splits) == 2: #country data
			countryName = splits[0]
			country2digit = splits[1]
		else: #people data
			personName = splits[0]
			personType = splits[1]
			country2digit = splits[2]			
		print '%s^%s^%s^%s' % (country2digit,personType,personName,countryName)
	except: #errors are going to make your job fail which you may or may not want

Don’t forget:

chmod a+x smplMapper.py

Great! We just took care of #1 but time to test and see what is going to the reducer.

From the command line run:

cat customers.dat countries.dat|./smplMapper.py|sort

Which will result in:

CA^not so good^Yo Yo Ma^-1
CA^valued^Jon Sneed^-1
CA^valued^Jon York^-1
CA^valued^Sam Sneed^-1
JA^not so bad^Jim Davis^-1
UK^-1^-1^United Kingdom
UK^not so good^Arnold Wesise^-1
UK^valued^Alex Ball^-1
US^-1^-1^United States
US^not bad^Alice Bob^-1
US^not bad^Henry Bob^-1

Notice how this is sorted so the country is first and the people in that country after it (so we can grab the correct country name as we loop) and with the type of customer also sorted (but within country) so we can properly count the types within the country. =8^)

Let us hold off on #2 for a moment (just hang in there it will all come together soon I promise) and get smplReducer.py working first.

#!/usr/bin/env python
import sys
# maps words to their counts
foundKey = ""
foundValue = ""
isFirst = 1
currentCount = 0
currentCountry2digit = "-1"
currentCountryName = "-1"
isCountryMappingLine = False

# 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
		country2digit,personType,personName,countryName = line.split('^')
		#the first line should be a mapping line, otherwise we need to set the currentCountryName to not known
		if personName == "-1": #this is a new country which may or may not have people in it
			currentCountryName = countryName
			currentCountry2digit = country2digit
			isCountryMappingLine = True
			isCountryMappingLine = False # this is a person we want to count
		if not isCountryMappingLine: #we only want to count people but use the country line to get the right name 

			#first check to see if the 2digit country info matches up, might be unkown country
			if currentCountry2digit != country2digit:
				currentCountry2digit = country2digit
				currentCountryName = '%s - Unkown Country' % currentCountry2digit
			currentKey = '%s\t%s' % (currentCountryName,personType) 
			if foundKey != currentKey: #new combo of keys to count
				if isFirst == 0:
					print '%s\t%s' % (foundKey,currentCount)
					currentCount = 0 #reset the count
					isFirst = 0
				foundKey = currentKey #make the found key what we see so when we loop again can see if we increment or print out
			currentCount += 1 # we increment anything not in the map list

	print '%s\t%s' % (foundKey,currentCount)

Don’t forget:

chmod a+x smplReducer.py

And then run:

cat customers.dat countries.dat|./smplMapper.py|sort|./smplReducer.py

And voila!

Canada	not so good	1
Canada	valued	3
JA - Unkown Country	not so bad	1
United Kingdom	not so good	1
United Kingdom	valued	1
United States	not bad	2

So now #3 and #4 are done but what about #2? 

First put the files into Hadoop:

hadoop fs -put ~/mayo/customers.dat .
hadoop fs -put ~/mayo/countries.dat .

And now run it like this (assuming you are running as hadoop in the bin directory):

hadoop jar ../contrib/streaming/hadoop-0.20.1+169.89-streaming.jar -D mapred.reduce.tasks=4 -file ~/mayo/smplMapper.py -mapper smplMapper.py -file ~/mayo/smplReducer.py -reducer smplReducer.py -input customers.dat -input countries.dat -output mayo -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner -jobconf stream.map.output.field.separator=^ -jobconf stream.num.map.output.key.fields=4 -jobconf map.output.key.field.separator=^ -jobconf num.key.fields.for.partition=1

Let us look at what we did:

hadoop fs -cat mayo/part*

Which results in:

Canada	not so good	1
Canada	valued	3
United Kingdom	not so good	1
United Kingdom	valued	1
United States	not bad	2
JA - Unkown Country	not so bad	1

So #2 is the partioner KeyFieldBasedPartitioner explained here further Hadoop Wiki On Streaming which allows the key to be whatever set of columns you output (in our case by country) configurable by the command line options and the rest of the values are sorted within that key and sent to the reducer together by key.

And there you go … Simple Python Scripting Implementing Streaming in Hadoop.  

Grab the tar here and give it a spin.

Joe Stein
Twitter: @allthingshadoop
Connect: On Linked In

Categories: Hadoop, MapReduce, Python
  1. Ganesh
    March 8, 2012 at 8:26 am


    I am doing a similar thing except that I want to run a binary executable from the mapper function. I had passed the binary through the -file option but the python code is not able to access the executable during the streaming job. Any idea where to copy the executable so the python code can access it during execution.


  2. Chris
    October 4, 2012 at 8:53 am

    You need to re-edit your slides and stuff… The links to this article refer to 2010/12/16. Took me a bit to search for your post.

  3. Thanks ForYour Help
    January 3, 2013 at 3:19 pm


  4. Max
    May 26, 2013 at 5:24 am

    Hi, thanks for this guide.

    Can you help me with my sorting problem.

    When I do: cat absatz.txt tag.txt | ./joinAbsatzTagMapper.py | sort | less

    this is what comes out(the interestening reagion of the whole output(2 big 2 show)):


    what im trying to do is to join the data of 2 files by the key tagId.

    this is what my python script is basically doing:

    for line in sys.stdin:
    line = line.strip()
    data = line.split(‘,’)

    tagId = “-1”
    prodId = “-1”
    menge = “-1”
    jahr = “-1”

    if len(data[1]) == 10: # we are in tag.txt
    tagId = data[0]
    jahr = data[2]

    else: # we are in absatz.txt
    prodId = data[0]
    menge = data[2]
    tagId = data[1]

    print ‘%s^%s^%s^%s’ % (tagId, prodId, menge, jahr)

    the file contents:

    tag.txt contains lots of data like:


    tagId,date,year <- this is what the coulmns stand for
    and absatz.txt contains data like:


    prodId,tagId,menge <- the columns
    sry that some idefntifiers are written in the german language, let me explain what they mean:
    absatz = sale
    tag = day
    menge = amount

  1. January 5, 2011 at 11:55 pm
  2. March 11, 2011 at 8:48 am
  3. June 28, 2011 at 2:16 am
  4. December 16, 2011 at 1:27 pm
  5. July 9, 2012 at 10:48 am
  6. May 28, 2013 at 9:38 pm

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out /  Change )

Facebook photo

You are commenting using your Facebook account. Log Out /  Change )

Connecting to %s

%d bloggers like this: