阅读目录

• 1. Python MapReduce 代码
• 2. 在Hadoop上运行python代码
• 3. 利用python的迭代器和生成器优化Mapper 和 Reducer代码
• 4. 参考

#!/usr/bin/env python
import sys
for line in sys.stdin:line = line.strip()words = line.split()for word in words:print "%s\t%s" % (word, 1)

chmod +x hadoop-0.20.2/test/code/mapper.py

#!/usr/bin/env python
from operator import itemgetter
import syscurrent_word = None
current_count = 0
word = Nonefor line in sys.stdin:line = line.strip()word, count = line.split('\t', 1)try:count = int(count)except ValueError:  #count如果不是数字的话，直接忽略掉continueif current_word == word:current_count += countelse:if current_word:print "%s\t%s" % (current_word, current_count)current_count = countcurrent_word = wordif word == current_word:  #不要忘记最后的输出print "%s\t%s" % (current_word, current_count)

chmod +x hadoop-0.20.2/test/code/reducer.py

str = 'server=mpilgrim&ip=10.10.10.10&port=8080'
print str.split('=', 1)[0]  #1表示=只截一次
print str.split('=', 1)[1]
print str.split('=')[0]
print str.split('=')[1]

bin/hadoop dfs -copyFromLocal /test/datas  hdfs_in

bin/hadoop dfs -ls

bin/hadoop dfs -ls /user/rte/hdfs_in

bin/hadoop jar contrib/streaming/hadoop-*streaming*.jar \
-file test/code/mapper.py     -mapper test/code/mapper.py \
-file test/code/reducer.py    -reducer test/code/reducer.py \
-input /user/rte/hdfs_in/*    -output /user/rte/hdfs_out

bin/hadoop dfs -ls /user/rte/hdfs_out

bin/hadoop dfs -cat /user/rte/hdfs_out2/part-00000

#!/usr/bin/env python
import sys
def read_input(file):for line in file:yield line.split()def main(separator='\t'):data = read_input(sys.stdin)for words in data:for word in words:print "%s%s%d" % (word, separator, 1)if __name__ == "__main__":main()

#!/usr/bin/env python
from operator import itemgetter
from itertools import groupby
import sysdef read_mapper_output(file, separator = '\t'):for line in file:yield line.rstrip().split(separator, 1)def main(separator = '\t'):data = read_mapper_output(sys.stdin, separator = separator)for current_word, group in groupby(data, itemgetter(0)):try:total_count = sum(int(count) for current_word, count in group)print "%s%s%d" % (current_word, separator, total_count)except valueError:passif __name__ == "__main__":main()

from itertools import groupby
from operator import itemgetterthings = [('2009-09-02', 11),('2009-09-02', 3),('2009-09-03', 10),('2009-09-03', 4),('2009-09-03', 22),('2009-09-06', 33)]sss = groupby(things, itemgetter(0))
for key, items in sss:print keyfor subitem in items:print subitemprint '-' * 20