我們知道MapReduce是hadoop這隻大象的核心,Hadoop中,資料處理核心就是 MapReduce 程式設計模型。一個Map/Reduce 通常會把輸入的資料集切分為若干獨立的資料區塊,由 map任務(task)以完全並行的方式處理它們。框架會對map的輸出先進行排序, 然後把結果輸入給reduce任務。通常作業的輸入和輸出都會儲存在檔案系統中。因此,我們的程式設計中心主要是 mapper階段和reducer階段。
下面來從零開發一個MapReduce程序,並在hadoop叢集上運行。
mapper程式碼map.py:
import sys for line in sys.stdin: word_list = line.strip().split(' ') for word in word_list: print '\t'.join([word.strip(), str(1)])
#View Code
reducer程式碼reduce.py:
import sys cur_word = None sum = 0 for line in sys.stdin: ss = line.strip().split('\t') if len(ss) < 2: continue word = ss[0].strip() count = ss[1].strip() if cur_word == None: cur_word = word if cur_word != word: print '\t'.join([cur_word, str(sum)]) cur_word = word sum = 0 sum += int(count) print '\t'.join([cur_word, str(sum)]) sum = 0
View Code
資源檔src.txt(測試用,在叢集中跑時,記得上傳到hdfs):
hello ni hao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni haoao ni haoni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao ni hao Dad would get out his mandolin and play for the family Dad loved to play the mandolin for his family he knew we enjoyed singing I had to mature into a man and have children of my own before I realized how much he had sacrificed I had to,mature into a man and,have children of my own before.I realized how much he had sacrificed
View Code
首先本地偵錯查看結果是否正確,輸入命令以下:
cat src.txt | python map.py | sort -k 1 | python reduce.py
命令列中輸出的結果:
a 2 and 2 and,have 1 ao 1 before 1 before.I 1 children 2 Dad 2 enjoyed 1 family 2 for 2 get 1 had 4 hao 33 haoao 1 haoni 3 have 1 he 3 hello 1 his 2 how 2 I 3 into 2 knew 1 loved 1 man 2 mandolin 2 mature 1 much 2 my 2 ni 34 of 2 out 1 own 2 play 2 realized 2 sacrificed 2 singing 1 the 2 to 2 to,mature 1 we 1 would 1
View Code
透過調試發現本地調試,程式碼是OK的。下面扔到集群上面跑。為了方便,專門寫了一個腳本 run.sh,解放勞動力嘛。
HADOOP_CMD="/home/hadoop/hadoop/bin/hadoop" STREAM_JAR_PATH="/home/hadoop/hadoop/contrib/streaming/hadoop-streaming-1.2.1.jar" INPUT_FILE_PATH="/home/input/src.txt" OUTPUT_PATH="/home/output" $HADOOP_CMD fs -rmr $OUTPUT_PATH $HADOOP_CMD jar $STREAM_JAR_PATH \ -input $INPUT_FILE_PATH \ -output $OUTPUT_PATH \ -mapper "python map.py" \ -reducer "python reduce.py" \ -file ./map.py \ -file ./reduce.py
下面解析下腳本:
HADOOP_CMD: hadoop的bin的路径 STREAM_JAR_PATH:streaming jar包的路径 INPUT_FILE_PATH:hadoop集群上的资源输入路径 OUTPUT_PATH:hadoop集群上的结果输出路径。(注意:这个目录不应该存在的,因此在脚本加了先删除这个目录。**注意****注意****注意**:若是第一次执行,没有这个目录,会报错的。可以先手动新建一个新的output目录。) $HADOOP_CMD fs -rmr $OUTPUT_PATH $HADOOP_CMD jar $STREAM_JAR_PATH \ -input $INPUT_FILE_PATH \ -output $OUTPUT_PATH \ -mapper "python map.py" \ -reducer "python reduce.py" \ -file ./map.py \ -file ./reduce.py #这里固定格式,指定输入,输出的路径;指定mapper,reducer的文件; #并分发mapper,reducer角色的我们用户写的代码文件,因为集群其他的节点还没有mapper、reducer的可执行文件。
# 輸入以下指令查看經過reduce階段後輸出的記錄:
cat src.txt | python map.py | sort -k 1 | python reduce.py | wc -l 命令行中输出:43
在瀏覽器輸入:master:50030 查看任務的詳細情況。
Kind % Complete Num Tasks Pending Running Complete Killed Failed/Killed Task Attempts map 100.00% 2 0 0 2 0 0 / 0 reduce 100.00% 1 0 0 1 0 0 / 0
Map-Reduce Framework中看到這個。
Counter Map Reduce Total Reduce output records 0 0 43
證明整個過程成功。第一個hadoop程式開發結束。
以上是Python開發MapReduce系列之WordCount Demo的詳細內容。更多資訊請關注PHP中文網其他相關文章!