用python寫MapReduce函數——以WordCount為例

      盡管Hadoop框架是用java寫的，但是Hadoop程序不限於java，可以用python、C++、ruby等。本例子中直接用python寫一個MapReduce實例，而不是用Jython把python代碼轉化成jar文件。

      例子的目的是統計輸入文件的單詞的詞頻。

• 輸入：文本文件
• 輸出：文本（每行包括單詞和單詞的詞頻，兩者之間用'\t'隔開）

 

1. Python MapReduce 代碼

      使用python寫MapReduce的“訣竅”是利用Hadoop流的API，通過STDIN(標准輸入)、STDOUT(標准輸出)在Map函數和Reduce函數之間傳遞數據。

      我們唯一需要做的是利用Python的sys.stdin讀取輸入數據，並把我們的輸出傳送給sys.stdout。Hadoop流將會幫助我們處理別的任何事情。

1.1 Map階段：mapper.py

在這里，我們假設把文件保存到hadoop-0.20.2/test/code/mapper.py

#!/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)

文件從STDIN讀取文件。把單詞切開，並把單詞和詞頻輸出STDOUT。Map腳本不會計算單詞的總數，而是輸出<word> 1。在我們的例子中，我們讓隨后的Reduce階段做統計工作。

為了是腳本可執行，增加mapper.py的可執行權限

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

1.2 Reduce階段：reducer.py

在這里，我們假設把文件保存到hadoop-0.20.2/test/code/reducer.py

#!/usr/bin/env python
from operator import itemgetter
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:  #count如果不是數字的話，直接忽略掉 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 word == current_word:  #不要忘記最后的輸出 print "%s\t%s" % (current_word, current_count)

文件會讀取mapper.py 的結果作為reducer.py 的輸入，並統計每個單詞出現的總的次數，把最終的結果輸出到STDOUT。

為了是腳本可執行，增加reducer.py的可執行權限

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

細節：split(chara, m)，第二個參數的作用，下面的例子很給力

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]

輸出

server
mpilgrim&ip=10.10.10.10&port=8080
server
mpilgrim&ip　

1.3 測試代碼（cat data | map | sort | reduce）

這里建議大家在提交給MapReduce job之前在本地測試mapper.py 和reducer.py腳本。否則jobs可能會成功執行，但是結果並非自己想要的。

功能性測試mapper.py 和 reducer.py

[rte@hadoop-0.20.2]$cd test/code
[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py
foo	1
foo	1
quux	1
labs	1
foo	1
bar	1
quux	1
[rte@code]$echo "foo foo quux labs foo bar quux" | ./mapper.py | sort -k1,1 | ./reducer.py
bar	1
foo	3
labs	1
quux	2

 細節：sort -k1,1  參數何意？

-k, -key=POS1[,POS2]     鍵以pos1開始，以pos2結束

有時候經常使用sort來排序，需要預處理把需要排序的field語言在最前面。實際上這是

完全沒有必要的，利用-k參數就足夠了。

比如sort all

1 4
2 3
3 2
4 1
5 0

如果sort -k 2的話，那么執行結果就是

5 0
4 1
3 2
2 3
1 4

 

2. 在Hadoop上運行python代碼

2.1 數據准備

下載以下三個文件的

• Plain Text UTF-8
• Plain Text UTF-8
• Plain Text UTF-8

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

2.2 運行

把本地的數據文件拷貝到分布式文件系統HDFS中。

bin/hadoop dfs -copyFromLocal /test/datas  hdfs_in

查看

bin/hadoop dfs -ls

結果

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

查看具體的文件

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

執行MapReduce job

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

實例輸出

查看輸出結果是否在目標目錄/user/rte/hdfs_out

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

輸出

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

查看結果

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

輸出

以上已經達成目的了，但是可以利用python迭代器和生成器優化

 

3. 利用python的迭代器和生成器優化Mapper 和 Reducer代碼

3.1 python中的迭代器和生成器

   看這

3.2 優化Mapper 和 Reducer代碼

mapper.py

#!/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()

reducer.py

#!/usr/bin/env python
from operator import itemgetter
from itertools import groupby
import sys

def 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:
            pass

if __name__ == "__main__":
    main()

細節：groupby

from itertools import groupby
from operator import itemgetter

things = [('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 key
    for subitem in items:
        print subitem
    print '-' * 20

結果

>>> 
2009-09-02
('2009-09-02', 11)
('2009-09-02', 3)
--------------------
2009-09-03
('2009-09-03', 10)
('2009-09-03', 4)
('2009-09-03', 22)
--------------------
2009-09-06
('2009-09-06', 33)
--------------------

注 

• groupby(things, itemgetter(0)) 以第0列為排序目標
• groupby(things, itemgetter(1))以第1列為排序目標
• groupby(things)以整行為排序目標

4. 參考

python中的split函數中的參數問題

Writing an Hadoop MapReduce Program in Python

shell的sort命令的-k參數