Python模拟MapReduce的流程

linghaiyan

　　说一下开发环境，是在WIN7下面进行的。
　　python环境是：python2.7
　　MapReduce的主要流程有：
　　Map阶段->Shuffle阶段->Reduce阶段。
　　那么一下分别对应三个python脚本片段：
　　数据样本：test.txt

a b c d
a b c d
aa bb cc dd
ee ff gg hh
foo foo quux labs foo bar quux
　　第一个脚本，叫做mapper.py

#!/bin/env python
import sys
def mapword(w):
print "%s\t%d"%(w,1)
for line in sys.stdin:
line = line.strip()
words = line.split()
m = map(mapword,words)
　　第二个脚本，叫做sort.py

import sys
def mapword(w):
print "%s\t%d"%(w,1)
m = []
for line in sys.stdin:
line = line.strip()
word,count = line.split('\t')
m.append((word,count))
m = sorted(m)
for i in m:
print "%s\t%s"%i
　　注意，实际上在分布式环境下，这个是sort by而不是order by。
　　也就是说在每个分区下是有序的，而并不是全局有序。
　　如果要做到全局有序，必须在一个reduce里面进行。
　　第三脚本，是reducer.py

import sys
current_word = None
current_count = 0
word = None
for line in sys.stdin:
line = line.strip()
word, count = line.split('\t', 1)
try:
count = int(count)
except ValueError:
continue
if current_word == word:
current_count += count
else:
if current_word:
print '%s\t%s' % (current_word, current_count)
current_count = count
current_word = word
if current_word == word:
print '%s\t%s' % (current_word, current_count)

　　在这个简单的word count例子里面。
　　ruducer的作用就像是一个dict或者叫做map，将数据维护到一份内存里面。然后在最后一起输出。
　　在Windows下的调试命令如下：
　　>type test.txt | python mapper.py | python sort.py | python reducer.py
　　结果如下：