与这篇博文有点关系的可以参考下:
① MapReduce 计算框架 —— 执行流程详解
② 在Linux环境实现wordcount:mapper,reducer的代码创建,脚本实现map,reduce
③ Linux实现 map 返回列表形式操作
④ Linux hadoop 脚本实现 reduce合并数据
前提:集群已经搭建好hadoop,启动集群。
cd /usr/hadoop/hadoop-2.7.3/
sbin/start-all.sh
首先我们需要有数据文档,1.data、The_Man_of_Property.txt。
head -3 1.data
创建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
结果显示如下:
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
结果如下:
通过正则表达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
结果如下:
这个结果才是我们想要的!!!
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)))
# 创建文件
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
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
对wordcount有更一步的理解,为什么不用IDE、pycharm呢? 这是考虑到企业以前员工会在服务器进行需求操作,因此在节点进行操作锻炼。
一些小知识点:
查找文件:* --> 匹配任何字符
find /usr/hadoop -name "hadoop-streaming*"
需求:sh脚本内容有streaming,要查有关于streaming在脚本的内容
find ./ -name "*.sh" | xargs grep "streaming"
history 可以显示自己的历史输入命令
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://blog.csdn.net/weixin_44775255/article/details/120912957
内容来源于网络,如有侵权,请联系作者删除!