hadoop是建立在MapReduce機制之上，其中wordcount是hadoop最典型的一個例項，然而眾所周知，hadoop的原始碼是java，並且大多數的hadoop程式碼都是基於java搭建起來，那如何利用python實現wordcount，這將是本篇部落格主要想完成的功能，並將寫好的程式放入hadoop叢集上跑

新建mapper.py

#-*- encoding=UTF-8 -*-

import sys
import re
##標準輸入
for line in sys.stdin:
    line = line.strip()
    words = re.split
 
('，',line)
    for word in words:
        print("{0}\t{1}".format(word,1))

這裡主要利用sys.stdin進行輸入,sys.stdout進行輸出，其中print為map到reduce這一段充當了標準輸出這一角色

輸入檔案 input.txt

hello,liming
hi,zhangsan
haha,hehe,liming
wangmazi,map
hadoop,hdfs,hbase
map,reduce,reduce

map測試

輸入以下指令對mapper的程式進行測試

cat input.txt
 
 | python mapper.py

輸出結果如下

hello   1
liming  1
hi  1
zhangsan    1
haha    1
hehe    1
liming  1
wangmazi    1
map 1
hadoop  1
hdfs    1
hbase   1
map 1
reduce  1
reduce  1

通過上述結果，發現map將單詞進行了分割，每個單詞都對應著自己的一個出現次數，接下來，基於hadoop的機制會將這些單詞進行排序，然後再傳給reduce進行處理

編寫reducer.py

# -*- encoding=UTF-8 -*-
 

import sys

cur_word = None
cur_count = 0
word = None

for line in sys.stdin:
    word,count = line.split('\t',1)

    count = int(count)

    if cur_word == word:
        cur_count += count
    else:
        if cur_word:
            print("{0}\t{1}".format(cur_word,cur_count))
        cur_word = word
        cur_count = count

# 最後一組的輸出
if word:
    print("{0}\t{1}".format(cur_word, cur_count))

reducer.py的編寫主要基於map排序過後進行，這是由於到將map的資料結果放到hdfs中時，會進行排序

測試reducer.py

輸入以下命令進行測試

cat input.txt | python mapper.py | sort | python reducer.py 

測試結果如下

hadoop  1
haha    1
hbase   1
hdfs    1
hehe    1
hello   1
hi  1
liming  2
map 2
reduce  2
wangmazi    1
zhangsan    1

至此，wordcount的map和reduce程式完全寫完，下面將把程式上傳至hadoop叢集上跑

將input.txt上傳

hadoop fs -put input.txt *** (hdfs目錄下)

shell指令碼

一般來說，都是通過指令的方式進行，但是指令會過長，因此寫一個簡單的shell指令碼便可完成

#!/bin/bash

hadoop fs -rm -r -f ***/wordcount
hadoop jar ***/hadoop-mapreduce/hadoop-streaming.jar \
-libjars  *** \
-jobconf mapreduce.reduce.shuffle.memory.limit.percent=0.1 \
-jobconf mapreduce.reduce.shuffle.input.buffer.percent=0.1 \
-jobconf mapred.map.capacity=100 \
-jobconf mapreduce.reduce.memory.mb=8182 \
-jobconf mapreduce.reduce.java.opts=-Xms1600m \
-jobconf mapred.map.capacity=100 \
-jobconf mapred.reduce.capacity=100 \
-jobconf mapred.reduce.tasks=600 \
-jobconf mapreduce.job.queuename=root.default \
-jobconf mapreduce.map.cpu.vcores=2 \
-jobconf mapreduce.reduce.cpu.vcores=4 \
-jobconf mapred.job.name=zds_sub_model_score \
-file mapper.py \
-file reducer.py \
-mapper "python mapper.py" \
-reducer "python reducer.py" \
-input  ***/input.txt \
-output ***/wordcount \

指令碼中

• *依據自己機器上的目錄進行設定
• 第3行刪除已有的wordcount檔案
• 第4、5行都是指定的jar包，依機器而定
• 第6-17行指定了各項引數
• 第18、19行指定了釋出的程式檔案
• 第20、21行指定執行的檔案
• 第22行為輸入檔案
• 第23行為輸出檔案，即是wordcount的輸出

將wordcount拉下來

hadoop fs -getmerge ***/wordcount wordcount

wordcount結果

hadoop  1
haha    1
hbase   1
hdfs    1
hehe    1
hello   1
hi  1
liming  2
map 2
reduce  2
wangmazi    1
zhangsan    1

發現在hadoop上執行的wordcount的結果和本地執行的結果一樣，便驗證了本文的方法