Python模拟MapReduce的流程

说一下开发环境，是在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

 

结果如下：