文章目录
前提：集群已经搭建好hadoop，启动集群。
cd /usr/hadoop/hadoop-2.7.3/
sbin/start-all.sh
1、数据准备
首先我们需要有数据文档，1.data、The_Man_of_Property.txt。
head -3 1.data
2、map创建的初始
创建map.py，vi map.py
import sys
import time
import re
p = re.compile(r'\w+')
for line in sys.stdin: #读取一行数据，标准输入读取
ss = line.strip().split(' ') #读取每行数据以空格划分，strip()注释掉首尾是空格或\n。
for s in ss: #对每个单词进行操作
time.sleep(1) #用来测试，方便查看
#s = p.findall(s)[0].lower()
if s.strip() != "": # 读取的不是空格
print(type(ss))
#print("%s\t%s" % (s,1))
print(ss) # 输出数据
我们需要的map结果是需要统计次数的；
因此输出要改为：print("%s\t%s" % (s,1))，注释掉print(type(ss)) 。
执行：cat 1.data | python map.py 结果显示如下：
3、在map.py基础下创建red.py
import sys
current_word = None
sum = 0 #初始化
for line in sys.stdin:
word, val = line.strip().split("\t") # 对map输出结果作为标准输入，读取一行
if current_word == None: #当前单词为None
current_word = word #就要把map出来的单词赋给current_word
if current_word != word: #这个单词与上一个不一致
print("%s\t%s" % (current_word, sum)) #把上一个单词sum输出
current_word = word #把下一个单词赋给当前单词
sum = 0 #重新进行计算次数
sum += int(val) #对相同单词，次数+1
print("%s\t%s" % (current_word, str(sum)))
一个例子：
map输出：
current_word = None
sum = 0
is 1 cur=is word=is sum=0+1=1 #执行sum += int(val)
is 1 cur=is word=is sum=1+1=2 #执行sum += int(val)
is 1 cur=is word=today sum=2+1=3 => is 3 #执行sum += int(val)
today 1 cur=today word=good sum=0+1=1 => today 1 #执行if current_word != word
good 1 cur=good sum=0+1=1 => goog 1 #执行if current_word != word
进行测试map，red，先注释掉time.sleep(1)
执行：cat 1.data | python map.py | sort -k1 | python red.py 结果如下：
4、map.py 重修版
通过正则表达findall(s)，消除字符。
这个存在符号，不符合我们想要的结果。重新编辑map.py
import sys
import time
import re
p = re.compile(r'\w+')
for line in sys.stdin:
ss = line.strip().split(' ')
for s in ss:
#time.sleep(1)
if len(p.findall(s)) < 1: #符号的大小小于1
continue #过滤掉符号
s = p.findall(s)[0].lower() # 正则匹配，转为小写
if s.strip() != "":
print ("%s\t%s" % (s, 1))
#print(ss)
#print(type(ss))
修改完map.py，执行cat 1.data | python map.py | sort -k1 | python red.py 结果如下：
这个结果才是我们想要的！！！
5、正确完整代码：
map.py，不能有#注释解释。
import sys
import time
import re
p = re.compile(r'\w+')
for line in sys.stdin:
ss = line.strip().split(' ')
for s in ss:
#time.sleep(1)
if len(p.findall(s)) < 1:
continue
s = p.findall(s)[0].lower()
if s.strip() != "":
print ("%s\t%s" % (s, 1))
#print ss
#print(type(ss))
red.py 代码， 不能有#注释解释。
import sys
current_word = None
sum = 0
for line in sys.stdin:
word, val = line.strip().split("\t")
if current_word == None:
current_word = word
if current_word != word:
print("%s\t%s" % (current_word, sum))
current_word = word
sum = 0
sum += int(val)
print("%s\t%s" % (current_word, str(sum)))
6、用脚本run.sh 一步到位 执行map.py与red.py
6.1 上传数据文档到hdfs
# 创建文件
hadoop fs -mkdir /test
hadoop fs -mkdir /output
# 上传文件
fs -put ./1.data /test/
fs -put ./The_Man_of_Property.txt /test/
# 查看文件
hadoop fs -ls /test
6.2 创建 run.sh脚本
vi run.sh 内容如下：
HADOOP_CMD="/usr/hadoop/hadoop-2.7.3/bin/hadoop"
STREAM_JAR_PATH="/usr/hadoop/hadoop-2.7.3/share/hadoop/tools/lib/hadoop-streaming-2.7.3.jar"
# hdfs输入路径
INPUT_FILE_PATH_1="/test/The_Man_of_Property.txt"
# hdfs输出路径
OUTPUT_PATH="/output/file_broadcast"
# 已经有输出路径就进行删除
# $HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH
# Step 1.
$HADOOP_CMD jar $STREAM_JAR_PATH \
-input $INPUT_FILE_PATH_1 \
-output $OUTPUT_PATH \
-mapper "python map.py" \
-reducer "python red.py" \
-file ./map.py \
-file ./red.py
脚本执行：sh -x run.sh
可以看到输出路径已经有文件了，我们注释掉 # $HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH。再执行 sh -x run.sh
由上图已经知道map，reduce 已经执行成功，证明这个run.sh 脚步已经一步到位。查看新的数据结果。
#查看输出路径是否已经创建 OUTPUT_PATH
hadoop fs -ls /output
hadoop fs -ls /output/file_broadcast
# cat输出最后10个数据
fs -cat /output/file_broadcast/part* | tail
7、总结：
对wordcount有更一步的理解，为什么不用IDE、pycharm呢？ 这是考虑到企业以前员工会在服务器进行需求操作，因此在节点进行操作锻炼。
一些小知识点：
查找文件：* --> 匹配任何字符
find /usr/hadoop -name "hadoop-streaming*"
需求：sh脚本内容有streaming，要查有关于streaming在脚本的内容
find ./ -name "*.sh" | xargs grep "streaming"
history 可以显示自己的历史输入命令
