在Hadoop上執行Python指令碼
之前已經配置好了Hadoop以及Yarn,可那只是第一步。下面還要在上面執行各種程式,這才是最重要的。
Ubuntu安裝時預設已經安裝了Python, 可以通過Python –version 查詢其版本。
因此我們可以直接執行python的指令碼了。
Python MapReduce Code
這裡我們要用到 Hadoop Streaming API, 通過STIDN(Standard input)和 STDOUT(Standard output)來向Map程式碼、Reduce程式碼傳遞資料。
Python有sys.stdin可以直接讀取資料,sys.stdout來輸出資料。
1 . 首先建立mapper.py.
用VIM建立mapper.py, 將檔案存在/home/hadoop路徑下, 程式碼如下:
#!/usr/bin/env python
import sys
# input comes from STDIN (standard input)
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# split the line into words
words = line.split()
# increase counters
for word in words:
# write the results to STDOUT (standard output);
# what we output here will be the input for the
# Reduce step, i.e. the input for reducer.py
#
# tab-delimited; the trivial word count is 1
print '%s\t%s' % (word, 1)
注意,儲存時存為unix編碼的,可以參考另一篇文章:
檔案儲存後,請注意將其許可權作出相應修改:
chmod a+x /home/hadoop/mapper.py
2 . 建立reduce.py
用VIM建立reduce.py, 將檔案存在/home/hadoop路徑下, 程式碼如下:
#!/usr/bin/env python
from operator import itemgetter
import sys
current_word = None
current_count = 0
word = None
# input comes from STDIN
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# parse the input we got from mapper.py
word, count = line.split('\t', 1)
# convert count (currently a string) to int
try:
count = int(count)
except ValueError:
# count was not a number, so silently
# ignore/discard this line
continue
# this IF-switch only works because Hadoop sorts map output
# by key (here: word) before it is passed to the reducer
if current_word == word:
current_count += count
else:
if current_word:
# write result to STDOUT
print '%s\t%s' % (current_word, current_count)
current_count = count
current_word = word
# do not forget to output the last word if needed!
if current_word == word:
print '%s\t%s' % (current_word, current_count)
檔案儲存後,請注意將其許可權作出相應修改:
chmod a+x /home/hadoop/reduce.py
首先可以在本機上測試以上程式碼,這樣如果有問題可以及時發現:
~$ echo "foo foo quux labs foo bar quux" | /home/hduser/mapper.py
執行結果如下:
再執行以下包含reducer.py的程式碼:
~$ echo "foo foo quux labs foo bar quux" | /home/hduser/mapper.py | sort -k1,1 | /home/hduser/reducer.py
結果如下:
在Hadoop上執行Python程式碼
準備工作:
下載文字檔案:
~$ mkdir tmp/guteberg
cd tmp/guteberg
wget http://www.gutenberg.org/files/5000/5000-8.txt
wget http://www.gutenberg.org/cache/epub/20417/pg20417.txt
然後把這二本書上傳到hdfs檔案系統上:
$ hdfs dfs -mkdir /user/input # 在hdfs上的該使用者目錄下建立一個輸入檔案的資料夾
$ hdfs dfs -put /home/hadoop/tmp/gutenberg/*.txt /user/input # 上傳文件到hdfs上的輸入資料夾中
尋找你的streaming的jar檔案存放地址,注意2.6的版本放到share目錄下了,可以進入hadoop安裝目錄尋找該檔案:
$ cd $HADOOP_HOME
$ find ./ -name "*streaming*.jar"
然後就會找到我們的share資料夾中的hadoop-straming*.jar檔案:
由於這個檔案的路徑比較長,因此我們可以將它寫入到環境變數:
vi ~/.bashrc # 開啟環境變數配置檔案
# 在裡面寫入streaming路徑
export STREAM=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar
由於通過streaming介面執行的指令碼太長了,因此直接建立一個shell名稱為run.sh來執行:
hadoop jar $STREAM \
-files /home/hadoop/mapper.py, /home/hadoop/reducer.py \
-mapper /home/hadoop/mapper.py \
-reducer /home/hadoop/reducer.py \
-input /user/input/*.txt \
-output /user/output
然後”source run.sh”來執行mapreduce。結果就響噹噹的出來啦。
用cat來看一下輸出結果如下: