Python进行简单的MapReduce(1)

所有操作，假定hadoop集群已经正常部署。
Python源码
mapper.py

#!/usr/bin python
 
import sys
 
# input comes from STDIN (standard input)
for line in sys.stdin:
    line = line.strip()
    words = line.split()
    for word in words:
        print '%s\\t%s' % (word, 1)

reduce.py

#!/usr/bin python
 
from operator import itemgetter
import sys

word2count = {}
 
# input comes from STDIN
for line in sys.stdin:
    line = line.strip()
 
    word, count = line.split('\\t', 1)
    try:
        count = int(count)
        word2count[word] = word2count.get(word, 0) + count
    except ValueError:
        # count was not a number, so silently
        # ignore/discard this line
        pass

sorted_word2count = sorted(word2count.items(), key=itemgetter(0))

for word, count in sorted_word2count:
    print '%s\\t%s'% (word, count)

先后存储在/home/src下，然后，cd到此目录
在hdfs上建立测试目录：
ls
hadoop fs -ls /user/hdfs
mkdir
hadoop fs -mkdir /user/hdfs/test

从本地磁盘copy测试文件到hdfs
hadoop fs -copuFromLocal /home/src/*.txt /user/hdfs/test/

使用streaming.jar执行mapreduce任务
hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar -mapper mapper.py -reducer reducer.py -file mapper.py -file reducer.py -input /user/hdfs/test/* -output /user/hdfs/test/reducer -mapper cat -reducer aggregate
执行结果：
......
14/11/26 12:54:52 INFO mapreduce.Job:  map 0% reduce 0%
14/11/26 12:54:59 INFO mapreduce.Job:  map 100% reduce 0%
14/11/26 12:55:04 INFO mapreduce.Job:  map 100% reduce 100%
14/11/26 12:55:04 INFO mapreduce.Job: Job job_1415798121952_0179 completed successfully
......
14/11/26 12:55:04 INFO streaming.StreamJob: Output directory: /user/hdfs/test/reducer
......
查看执行结果集文件
hadoop fs -ls /user/hdfs/test
......
drwxr-xr-x   - root Hadoop          0 2014-11-26 12:55 /user/hdfs/test/reducer
......