讓python在hadoop上跑起來
來自： http://www.cnblogs.com/wing1995/p/hadoop.html
duang~好久沒有更新博客啦，原因很簡單，實習啦～好吧，我過來這邊上班表示覺得自己簡直弱爆了。第一周，配置環境；第二周，將數據可視化，包括學習了excel2013的一些高大上的技能，例如數據透視表和mappower繪制3d地圖，當然本來打算是在tkinter里面運用matplotlib制作一個交互式的圖表界面，然而，畫出來的圖簡直不是excel2013能比的，由于對界面和matplotlib研究的也不是很深，短時間是沒法研究出來，上周真是多災多難；現在，第三周，開始接觸hadoop，雖說大多數現在的hadoop上運行的程序都是java，但是經過一周的java初入門，我還是果斷的選擇在hadoop上面跑python，是的，python是一個深坑，請大家隨我入坑，跟著 教程 一起學習如何用python寫hadoop的mapreduce吧！
關于hadoop，建議大家在自己的linux上面跟著網上的教程搭建一次單節點和多節點的hadoop平臺，我這里演示的是直接登錄服務器，所以環境神馬的都是現成的。關于 mapreduce ，我是新手，只能從“分而治之”的角度來考慮，首先“map”也就是”分”——數據分割，然后“reduce”對"map"處理后的結果進一步的運算，這里給出的例子是一般的hadoop入門程序 “WordCount” ，就是首先寫一個map程序用來將輸入的字符串分割成單個的單詞，然后reduce這些單個的單詞，相同的單詞就對其進行計數，不同的單詞分別輸出，結果輸出每一個單詞出現的頻數。這就是我們的簡單程序的思想，讓我們玩玩～
注意：關于數據的輸入輸出是通過sys.stdin（系統標準輸入）和sys.stdout（系統標準輸出）來控制數據的讀入與輸出。所有的腳本執行之前都需要修改權限，否則沒有執行權限，例如下面的腳本創建之前使用“chmod +x mapper.py”
1.mapper.py
1 #!/usr/bin/env python 2 import sys 3 4 for line in sys.stdin: # 遍歷讀入數據的每一行 5 6 line = line.strip() # 將行尾行首的空格去除 7 words = line.split() #按空格將句子分割成單個單詞 8 for word in words: 9 print '%s\t%s' %(word, 1)
2.reducer.py
1 #!/usr/bin/env python
2
3 from operator import itemgetter
4 import sys
5
6 current_word = None # 為當前單詞
7 current_count = 0 # 當前單詞頻數
8 word = None
9
10 for line in sys.stdin:
11 words = line.strip() # 去除字符串首尾的空白字符
12 word, count = words.split('\t') # 按照制表符分隔單詞和數量
13
14 try:
15 count = int(count) # 將字符串類型的‘1’轉換為整型1
16 except ValueError:
17 continue
18
19 if current_word == word: # 如果當前的單詞等于讀入的單詞
20 current_count += count # 單詞頻數加1
21 else:
22 if current_word: # 如果當前的單詞不為空則打印其單詞和頻數
23 print '%s\t%s' %(current_word, current_count)
24 current_count = count # 否則將讀入的單詞賦值給當前單詞，且更新頻數
25 current_word = word
26
27 if current_word == word:
28 print '%s\t%s' %(current_word, current_count)
在shell中運行以下腳本，查看輸出結果：
1 echo "foo foo quux labs foo bar zoo zoo hying" | /home/wuying/mapper.py | sort -k 1,1 | /home/wuying/reducer.py 2 3 # echo是將后面“foo ****”字符串輸出，并利用管道符“|”將輸出數據作為mapper.py這個腳本的輸入數據，并將mapper.py的數據輸入到reducer.py中，其中參數sort -k 1,1是將reducer的輸出內容按照第一列的第一個字母的ASCII碼值進行升序排序
其實，我覺得后面這個reducer.py處理單詞頻數有點麻煩，將單詞存儲在字典里面，單詞作為‘key’，每一個單詞出現的頻數作為'value'，進而進行頻數統計感覺會更加高效一點。因此，改進腳本如下：
mapper_1.py
但是，貌似寫著寫著用了兩個循環，反而效率低了。關鍵是不太明白這里的current_word和current_count的作用，如果從字面上老看是當前存在的單詞，那么怎么和遍歷讀取的word和count相區別？
下面看一些腳本的輸出結果：
我們可以看到，上面同樣的輸入數據，同樣的shell換了不同的reducer，結果后者并沒有對數據進行排序，實在是費解～
讓Python代碼在hadoop上跑起來！
一、準備輸入數據
接下來，先下載三本書：
1 $ mkdir -p tmp/gutenberg 2 $ cd tmp/gutenberg 3 $ wget http://www.gutenberg.org/ebooks/20417.txt.utf-8 4 $ wget http://www.gutenberg.org/files/5000/5000-8.txt 5 $ wget http://www.gutenberg.org/ebooks/4300.txt.utf-8
然后把這三本書上傳到hdfs文件系統上：
1 $ hdfs dfs -mkdir /user/${whoami}/input # 在hdfs上的該用戶目錄下創建一個輸入文件的文件夾
2 $ hdfs dfs -put /home/wuying/tmp/gutenberg/*.txt /user/${whoami}/input # 上傳文檔到hdfs上的輸入文件夾中
尋找你的streaming的jar文件存放地址，注意2.6的版本放到share目錄下了，可以進入hadoop安裝目錄尋找該文件：
$ cd $HADOOP_HOME $ find ./ -name "*streaming*"
然后就會找到我們的share文件夾中的hadoop-straming*.jar文件:
尋找速度可能有點慢，因此你最好是根據自己的版本號到對應的目錄下去尋找這個streaming文件，由于這個文件的路徑比較長，因此我們可以將它寫入到環境變量：
$ vi ~/.bashrc # 打開環境變量配置文件在里面寫入streaming路徑
export STREAM=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar</pre>
由于通過streaming接口運行的腳本太長了，因此直接建立一個shell名稱為run.sh來運行：
1 hadoop jar $STREAM \ 2 -files ./mapper.py,./reducer.py \ 3 -mapper ./mapper.py \ 4 -reducer ./reducer.py \ 5 -input /user/$(whoami)/input/*.txt \ 6 -output /user/$(whoami)/output然后"source run.sh"來執行mapreduce。結果就響當當的出來啦。這里特別要提醒一下：
1、一定要把本地的輸入文件轉移到hdfs系統上面，否則無法識別你的input內容；
2、一定要有權限，一定要在你的hdfs系統下面建立你的個人文件夾否則就會被denied，是的，就是這兩個錯誤搞得我在服務器上面痛不欲生，四處問人的感覺真心不如自己清醒對待來的好；
3、如果你是第一次在服務器上面玩hadoop，建議在這之前請在自己的虛擬機或者linux系統上面配置好偽分布式然后入門hadoop來的比較不那么頭疼，之前我并不知道我在服務器上面運維沒有給我運行的權限，后來在自己的虛擬機里面運行一下example實例以及wordcount才找到自己的錯誤。
好啦，然后不出意外，就會complete啦，你就可以通過如下方式查看計數結果：
上面的字數計量大小可能你們跟我的不一樣，那是因為我換了另外一個文檔測試，所以不用著急哈。
再次，感謝以下文檔的支持：
人生漫漫，且走且珍惜，加油，all is well, just do it!
</div>