本文實(shí)例講解的是一般的hadoop入門程序“WordCount”，就是首先寫一個(gè)map程序用來將輸入的字符串分割成單個(gè)的單詞，然后reduce這些單個(gè)的單詞，相同的單詞就對(duì)其進(jìn)行計(jì)數(shù)，不同的單詞分別輸出，結(jié)果輸出每一個(gè)單詞出現(xiàn)的頻數(shù)。

　　注意：關(guān)于數(shù)據(jù)的輸入輸出是通過sys.stdin（系統(tǒng)標(biāo)準(zhǔn)輸入）和sys.stdout（系統(tǒng)標(biāo)準(zhǔn)輸出）來控制數(shù)據(jù)的讀入與輸出。所有的腳本執(zhí)行之前都需要修改權(quán)限，否則沒有執(zhí)行權(quán)限，例如下面的腳本創(chuàng)建之前使用“chmod +x mapper.py”

1.mapper.py

#!/usr/bin/env python
import sys

for line in sys.stdin: # 遍歷讀入數(shù)據(jù)的每一行
  
  line = line.strip() # 將行尾行首的空格去除
  words = line.split() #按空格將句子分割成單個(gè)單詞
  for word in words:
    print '%s\t%s' %(word, 1)

2.reducer.py

#!/usr/bin/env python

from operator import itemgetter
import sys

current_word = None # 為當(dāng)前單詞
current_count = 0 # 當(dāng)前單詞頻數(shù)
word = None

for line in sys.stdin:
  words = line.strip() # 去除字符串首尾的空白字符
  word, count = words.split('\t') # 按照制表符分隔單詞和數(shù)量
  
  try:
    count = int(count) # 將字符串類型的‘1'轉(zhuǎn)換為整型1
  except ValueError:
    continue

  if current_word == word: # 如果當(dāng)前的單詞等于讀入的單詞
    current_count += count # 單詞頻數(shù)加1
  else:
    if current_word: # 如果當(dāng)前的單詞不為空則打印其單詞和頻數(shù)
      print '%s\t%s' %(current_word, current_count) 
    current_count = count # 否則將讀入的單詞賦值給當(dāng)前單詞，且更新頻數(shù)
    current_word = word

if current_word == word:
  print '%s\t%s' %(current_word, current_count)

在shell中運(yùn)行以下腳本，查看輸出結(jié)果：

echo "foo foo quux labs foo bar zoo zoo hying" | /home/wuying/mapper.py | sort -k 1,1 | /home/wuying/reducer.py

# echo是將后面“foo ****”字符串輸出，并利用管道符“|”將輸出數(shù)據(jù)作為mapper.py這個(gè)腳本的輸入數(shù)據(jù)，并將mapper.py的數(shù)據(jù)輸入到reducer.py中，其中參數(shù)sort -k 1,1是將reducer的輸出內(nèi)容按照第一列的第一個(gè)字母的ASCII碼值進(jìn)行升序排序

其實(shí)，我覺得后面這個(gè)reducer.py處理單詞頻數(shù)有點(diǎn)麻煩，將單詞存儲(chǔ)在字典里面，單詞作為‘key'，每一個(gè)單詞出現(xiàn)的頻數(shù)作為'value'，進(jìn)而進(jìn)行頻數(shù)統(tǒng)計(jì)感覺會(huì)更加高效一點(diǎn)。因此，改進(jìn)腳本如下：

mapper_1.py

但是，貌似寫著寫著用了兩個(gè)循環(huán)，反而效率低了。關(guān)鍵是不太明白這里的current_word和current_count的作用，如果從字面上老看是當(dāng)前存在的單詞，那么怎么和遍歷讀取的word和count相區(qū)別？

下面看一些腳本的輸出結(jié)果：

我們可以看到，上面同樣的輸入數(shù)據(jù)，同樣的shell換了不同的reducer，結(jié)果后者并沒有對(duì)數(shù)據(jù)進(jìn)行排序，實(shí)在是費(fèi)解～

讓Python代碼在hadoop上跑起來！

一、準(zhǔn)備輸入數(shù)據(jù)

接下來，先下載三本書：

$ mkdir -p tmp/gutenberg
$ cd tmp/gutenberg
$ wget http://www.gutenberg.org/ebooks/20417.txt.utf-8
$ wget http://www.gutenberg.org/files/5000/5000-8.txt
$ wget http://www.gutenberg.org/ebooks/4300.txt.utf-8

 然后把這三本書上傳到hdfs文件系統(tǒng)上：

 $ hdfs dfs -mkdir /user/${whoami}/input # 在hdfs上的該用戶目錄下創(chuàng)建一個(gè)輸入文件的文件夾
 $ hdfs dfs -put /home/wuying/tmp/gutenberg/*.txt /user/${whoami}/input # 上傳文檔到hdfs上的輸入文件夾中

尋找你的streaming的jar文件存放地址，注意2.6的版本放到share目錄下了，可以進(jìn)入hadoop安裝目錄尋找該文件：

$ cd $HADOOP_HOME
$ find ./ -name "*streaming*"

然后就會(huì)找到我們的share文件夾中的hadoop-straming*.jar文件:

尋找速度可能有點(diǎn)慢，因此你最好是根據(jù)自己的版本號(hào)到對(duì)應(yīng)的目錄下去尋找這個(gè)streaming文件，由于這個(gè)文件的路徑比較長(zhǎng)，因此我們可以將它寫入到環(huán)境變量：

$ vi ~/.bashrc # 打開環(huán)境變量配置文件
# 在里面寫入streaming路徑
export STREAM=$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar

由于通過streaming接口運(yùn)行的腳本太長(zhǎng)了，因此直接建立一個(gè)shell名稱為run.sh來運(yùn)行：

hadoop jar $STREAM \
-files ./mapper.py,./reducer.py \
-mapper ./mapper.py \
-reducer ./reducer.py \
-input /user/$(whoami)/input/*.txt \
-output /user/$(whoami)/output

然后"source run.sh"來執(zhí)行mapreduce。結(jié)果就響當(dāng)當(dāng)?shù)某鰜砝病＿@里特別要提醒一下：

1、一定要把本地的輸入文件轉(zhuǎn)移到hdfs系統(tǒng)上面，否則無法識(shí)別你的input內(nèi)容；

2、一定要有權(quán)限，一定要在你的hdfs系統(tǒng)下面建立你的個(gè)人文件夾否則就會(huì)被denied，是的，就是這兩個(gè)錯(cuò)誤搞得我在服務(wù)器上面痛不欲生，四處問人的感覺真心不如自己清醒對(duì)待來的好；

3、如果你是第一次在服務(wù)器上面玩hadoop，建議在這之前請(qǐng)?jiān)谧约旱奶摂M機(jī)或者linux系統(tǒng)上面配置好偽分布式然后入門hadoop來的比較不那么頭疼，之前我并不知道我在服務(wù)器上面運(yùn)維沒有給我運(yùn)行的權(quán)限，后來在自己的虛擬機(jī)里面運(yùn)行一下example實(shí)例以及wordcount才找到自己的錯(cuò)誤。

好啦，然后不出意外，就會(huì)complete啦，你就可以通過如下方式查看計(jì)數(shù)結(jié)果：

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家學(xué)習(xí)python軟件編程有所幫助。

最新評(píng)論