hadoop之使用python进行streaming编程

创建测试数据test.txt并写入如下内容

foo    bar    quux
foo    libs    bar
zoo    zoo    hying

1、创建mapper分割数据并写入标准输出流

$vi mapper.py#!/usr/bin/python
import sysfor line in sys.stdin:  # 遍历读入数据的每一行line = line.strip()  # 将行尾行首的空格去除words = line.split()  #按空格将句子分割成单个单词for word in words:print('%s\t%s' %(word, 1))

2、输出流作为输入数据，创建reduce 对输出结果计算

$vi reducer.py#!/usr/bin/python
from operator import itemgetter
import syscurrent_word = None  # 为当前单词
current_count = 0  # 当前单词频数
word = Nonefor line in sys.stdin:words = line.strip()  # 去除字符串首尾的空白字符word, count = words.split('\t')  # 按照制表符分隔单词和数量try:count = int(count)  # 将字符串类型的‘1’转换为整型1except ValueError:continueif current_word == word:  # 如果当前的单词等于读入的单词current_count += count  # 单词频数加1else:if current_word:  # 如果当前的单词不为空则打印其单词和频数print('%s\t%s' %(current_word, current_count))  current_count = count  # 否则将读入的单词赋值给当前单词，且更新频数current_word = wordif current_word == word:print ('%s\t%s' %(current_word, current_count))

3、本地测试执行

$cat test.txt | ./mapper.py | sort -k 1,1 | ./reducer.py

4、上传测试文件到hadoop文件系统

$hdfs dfs -mkdir /input
$hdfs dfs -put ./test.txt /input

5、调用streaming接口运行

$hadoop jar {$HADOOP_HOME}/share/hadoop/tools/lib/hadoop-streaming-3.2.1.jar -files ./mapper.py,./reducer.py -mapper ./mapper.py -reducer ./reducer.py -input /input/test.txt -output /output

6、查看执行结果

$hdfs dfs -cat /output/part-00000