嘗試著用3台虛擬機搭建了偽分散式系統,完整的搭建步驟等熟悉了整個分散式框架之後再寫,今天寫一下用python寫wordcount程式(MapReduce任務)的具體步驟。 MapReduce任務以來HDFS存儲和Yarn資源調度,所以執行MapReduce之前要先啟動HDFS和Yarn。我們都知道M ...
嘗試著用3台虛擬機搭建了偽分散式系統,完整的搭建步驟等熟悉了整個分散式框架之後再寫,今天寫一下用python寫wordcount程式(MapReduce任務)的具體步驟。
MapReduce任務以來HDFS存儲和Yarn資源調度,所以執行MapReduce之前要先啟動HDFS和Yarn。我們都知道MapReduce分Map階段和Reduce階段,這就需要我們 自己寫Map階段的處理方法和Reduce階段的處理方法。
MapReduce也支持除Java之外的其他語言,但要依賴流處理包(hadoop-streaming-2.7.4.jar),處理包不需要自己下載,hadoop本身帶的就有,hadoop2.7的在hadoop-2.7.4/share/hadoop/tools/lib目錄下,知道它所在的目錄是因為只執行MapReduce命令的時候要指定hadoop-streaming-2.7.4.jar的位置。
接下來就是用python寫Map的處理邏輯和Reduce的處理邏輯。wordcount是詞頻統計,要處理的原文本文件要上傳到HDFS上,流程是原文本以流式方式傳到Map函數,Map函數處理之後把結果傳到Reduce函數,整個處理完後結果會保存在HDFS上,流式處理可以理解成文本一行一行的在原文件、Map函數、Reduce函數、結果文件之間流動處理。
原文本:
hello world hello hadoop hadoop nihao world hello mapreduce
Map方法代碼:
#!/usr/bin/python import sys
for line in sys.stdin: line = line.strip() words = line.split(' ') for word in words: print('%s\t%s'%(word,1))
Reduce方法代碼:
#!/usr/bin/python import sys current_count = 0 current_word = None for line in sys.stdin: line = line.strip() word, count = line.split('\t', 1) count = int(count) 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
代碼的邏輯都很簡單,從標準輸入按行讀取處理數據,每行處理完print列印。
先在shell上測試一下:
#cat word.txt | ./mapper.py | sort
hadoop 1 hadoop 1 hello 1 hello 1 hello 1 mapreduce 1 nihao 1 world 1 world 1
sort是行之間按單詞首字母排序,在MapReduce上sort過程hadoop會處理。
如果沒有sort,結果是這樣的:
#cat word.txt | ./mapper.py
hello 1 world 1 hello 1 hadoop 1 hadoop 1 nihao 1 world 1 hello 1 mapreduce 1
#cat word.txt | ./mapper.py | sort |./reducer.py
hadoop 2 hello 3 mapreduce 1 nihao 1
測試完沒問題後就可以用MapReduce來執行了。
輸入命令:
hadoop jar hadoop-streaming-2.7.4.jar \
-input /wordcount/word.txt \
-output /wordcount/out \
-mapper /home/hadoop/apps/hadoop-2.7.4/file/wordcount_python/mapper.py \
-reducer /home/hadoop/apps/hadoop-2.7.4/file/wordcount_python/reducer.py
命令解釋:
第一行是指明用到的streaming包的位置,第二行指明原文件在HDFS上的路徑,第三行是輸出結果在HDFS上的路徑,輸出路徑原來不能存在,已存在的話會報錯,第四行和第五行指明Map方法和Reduce方法程式路徑。
mapper.py和reduce.py需要加上執行許可權,chmod +x mapper.py,它們兩個py文件不用放在HDFS上,放在本地即可。
執行後就會開啟MapReduce任務,配置沒問題的話就不會報錯,執行完成後會在MapReduce上生成/wordcount/out目錄裡面有兩個文件:
第二個是結果文件,第一個文件可以看到所占空間為0,cat一下什麼也沒有,只是一個處理成功的標識。
以上就是python寫wordcount的具體步驟,如有錯誤 歡迎指正!