Hadoop Steaming开发之WordCount

简单的WordCount栗子--类似于编程语言中的hello world

1.shell脚本run.sh

HADOOP_CMD="/usr/local/src/hadoop-1.2.1/bin/hadoop"
STREAM_JAR_PATH="/usr/local/src/hadoop-1.2.1/contrib/streaming/hadoop-streaming-1.2.1.jar"

INPUT_FILE_PATH="/input.txt"
OUTPUT_PATH="/output"

$HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH    #如果输出文件目录存在就先删除

# Step 1.
$HADOOP_CMD jar $STREAM_JAR_PATH 
    -input $INPUT_FILE_PATH                #-input:指定输入文件hdfs路径,支持通配符*,支持指定多个文件或目录, 可多次使用
    -output $OUTPUT_PATH                   #-output:指定输出文件的hdfs路径, 路径不能存在,执行作业用户必须有创建该目录的权限,只能用一次
    -mapper "python map.py"                #自己写的mapper程序
    -reducer "python red.py"               #自己写的reducer程序
    -file ./map.py                         #-file:分发文件到计算节点,1.map和reduce的执行文件;2.map和reduce要用的输入文件.如果要用的文件太大,需要用hdfs存储,类似的配置有-cacheFile,-cacheArchive分别用于向计算节点分发hdfs文件和hdfs压缩文件
    -file ./red.py 
17     -jobconf mapred.job.name="WordCount"    #设置作业的名称

2.map.py

#!/usr/local/bin/python

import sys

for line in sys.stdin:
    ss = line.strip().split(' ')
    for s in ss:
        if s.strip() != "":
            print "%s	%s" % (s, 1)      #标准输出用	分割,会自动识别为key,value

3.red.py

#!/usr/local/bin/python

import sys

current_word = None
count_pool = []
sum = 0

for line in sys.stdin:
    word, val = line.strip().split('	')

    if current_word == None:
        current_word = word

    if current_word != word:
        for count in count_pool:
            sum += count
        print "%s	%s" % (current_word, sum)
        current_word = word
        count_pool = []
        sum = 0

    count_pool.append(int(val))

for count in count_pool:
    sum += count
print "%s	%s" % (current_word, str(sum))

启动hadoop集群后, 运行shell脚本: 

 以下是Tracking URL中的部分内容:

查看输出到hdfs的目录/output

运行以下命令查看结果中排名前十的单词