摘要: ​尽管Hadoop框架是用java写的,但是Hadoop程序不限于java,可以用python、C++、ruby等。本例子中直接用python写一个MapReduce实例,而不是用Jython把python代码转化成jar文件。

例子的目的是统计输入文件的单词的词频。

  • 输入:文本文件
  • 输出:文本(每行包括单词和单词的词频,两者之间用'\t'隔开)

Python MapReduce 代码

使用python写MapReduce的“诀窍”是利用Hadoop流的API,通过STDIN(标准输入)、STDOUT(标准输出)在Map函数和Reduce函数之间传递数据。
我们唯一需要做的是利用Python的sys.stdin读取输入数据,并把我们的输出传送给sys.stdout。Hadoop流将会帮助我们处理别的任何事情。

Map阶段:mapper.py

在这里,我们假设把文件保存到hadoop-0.20.2/test/code/mapper.py

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

文件从STDIN读取文件。把单词切开,并把单词和词频输出STDOUT。Map脚本不会计算单词的总数,而是输出 1。在我们的例子中,我们让随后的Reduce阶段做统计工作。

为了是脚本可执行,增加mapper.py的可执行权限

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

Reduce阶段:reducer.py

在这里,我们假设把文件保存到hadoop-0.20.2/test/code/reducer.py

 1#!/usr/bin/env python
 2from operator import itemgetter
 3import sys
 4
 5current_word = None
 6current_count = 0
 7word = None
 8
 9for line in sys.stdin:
10    line = line.strip()
11    word, count = line.split('\t', 1)
12    try:
13        count = int(count)
14    except ValueError:  #count如果不是数字的话,直接忽略掉
15        continue
16    if current_word == word:
17        current_count += count
18    else:
19        if current_word:
20            print "%s\t%s" % (current_word, current_count)
21        current_count = count
22        current_word = word
23
24if word == current_word:  #不要忘记最后的输出
25    print "%s\t%s" % (current_word, current_count)

文件会读取mapper.py 的结果作为reducer.py 的输入,并统计每个单词出现的总的次数,把最终的结果输出到STDOUT。

为了是脚本可执行,增加reducer.py的可执行权限

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

细节:split(chara, m),第二个参数的作用,下面的例子很给力:

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

输出:

1server  
2mpilgrim&ip=10.10.10.10&port=8080  
3server  
4mpilgrim&ip

测试代码(cat data | map | sort | reduce)

这里建议大家在提交给MapReduce job之前在本地测试mapper.py 和reducer.py脚本。否则jobs可能会成功执行,但是结果并非自己想要的。

功能性测试mapper.py 和 reducer.py

 1[rte@hadoop-0.20.2]$cd test/code
 2[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py
 3foo 1
 4foo 1
 5quux    1
 6labs    1
 7foo 1
 8bar 1
 9quux    1
10[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py | sort -k1,1 | ./reducer.py
11bar 1
12foo 3
13labs    1
14quux    2

细节:sort -k1,1 参数何意?

-k, -key=POS1[,POS2] 键以pos1开始,以pos2结束

有时候经常使用sort来排序,需要预处理把需要排序的field语言在最前面。实际上这是

完全没有必要的,利用-k参数就足够了。

比如sort all

11 4
22 3
33 2
44 1
55 0

如果sort -k 2的话,那么执行结果就是

15 0
24 1
33 2
42 3
51 4

在Hadoop上运行python代码

数据准备

下载以下三个文件

我把上面三个文件放到hadoop-0.20.2/test/datas/目录下

运行

把本地的数据文件拷贝到分布式文件系统HDFS中。

1bin/hadoop dfs -copyFromLocal /test/datas  hdfs_in

查看

1bin/hadoop dfs -ls

结果

1drwxr-xr-x   - rte supergroup          0 2014-07-05 15:40 /user/rte/hdfs_in

##查看具体的文件

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

执行MapReduce job

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

实例输出

image

查看输出结果是否在目标目录/user/rte/hdfs_out

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

输出:

1Found 2 items
2drwxr-xr-x   - rte supergroup          0 2014-07-05 20:51 /user/rte/hdfs_out2/_logs
3-rw-r--r--   2 rte supergroup     880829 2014-07-05 20:51 /user/rte/hdfs_out2/part-00000

查看结果:

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

输出
image

以上已经达成目的了,但是可以利用python迭代器和生成器优化

利用python的迭代器和生成器优化Mapper 和 Reducer代码

python中的迭代器和生成器

python迭代器和生成器

优化Mapper 和 Reducer代码

mapper.py

 1#!/usr/bin/env python
 2import sys
 3def read_input(file):
 4    for line in file:
 5        yield line.split()
 6
 7def main(separator='\t'):
 8    data = read_input(sys.stdin)
 9    for words in data:
10        for word in words:
11            print "%s%s%d" % (word, separator, 1)
12
13if __name__ == "__main__":
14    main()

reducer.py

 1#!/usr/bin/env python
 2from operator import itemgetter
 3from itertools import groupby
 4import sys
 5
 6def read_mapper_output(file, separator = '\t'):
 7    for line in file:
 8        yield line.rstrip().split(separator, 1)
 9
10def main(separator = '\t'):
11    data = read_mapper_output(sys.stdin, separator = separator)
12    for current_word, group in groupby(data, itemgetter(0)):
13        try:
14            total_count = sum(int(count) for current_word, count in group)
15            print "%s%s%d" % (current_word, separator, total_count)
16        except valueError:
17            pass
18
19if __name__ == "__main__":
20    main()

细节:groupby

 1from itertools import groupby
 2from operator import itemgetter
 3
 4things = [('2009-09-02', 11),
 5          ('2009-09-02', 3),
 6          ('2009-09-03', 10),
 7          ('2009-09-03', 4),
 8          ('2009-09-03', 22),
 9          ('2009-09-06', 33)]
10
11sss = groupby(things, itemgetter(0))
12for key, items in sss:
13    print key
14    for subitem in items:
15        print subitem
16    print '-' * 20

结果:

 1>>>
 22009-09-02
 3('2009-09-02', 11)
 4('2009-09-02', 3)
 5--------------------
 62009-09-03
 7('2009-09-03', 10)
 8('2009-09-03', 4)
 9('2009-09-03', 22)
10--------------------
112009-09-06
12('2009-09-06', 33)
13--------------------

注:

groupby(things, itemgetter(0)) 以第0列为排序目标

groupby(things, itemgetter(1))以第1列为排序目标

groupby(things)以整行为排序目标


微信公众号

潘建锋

关于版权和转载

本文由 潘建锋 创作,采用 署名 4.0 国际 (CC BY 4.0) 国际许可协议进行授权。
本站文章除注明转载/出处外,均为本站原创或翻译,转载时请务必署名,否则,本人将保留一切追究责任的权利。
署名 4.0 国际 (CC BY 4.0)

转载规范

标题:hadoop平台wordcount程序的python实现
作者:潘建锋
原文:HTTPS://strikefreedom.top/wordcout-in-hadoop-by-python

关于留言和评论

如果您对本文《hadoop平台wordcount程序的python实现》的内容有任何疑问、补充或纠错,欢迎在下面的评论系统中留言,与作者一起交流进步,谢谢!(~ ̄▽ ̄)~