Python初次实现MapReduce——WordCount

时间:2024-01-10 20:44:50

前言

Hadoop 本身是用 Java 开发的,所以之前的MapReduce代码小练都是由Java代码编写,但是通过Hadoop Streaming,我们可以使用任意语言来编写程序,让Hadoop 运行。

本文用Python语言实现了词频统计功能,最后通过Hadoop Streaming使其运行在Hadoop上。

Python写MapReduce代码

使用Python写MapReduce的“诀窍”是利用Hadoop流的API,通过STDIN(标准输入)、STDOUT(标准输出)在Map函数和Reduce函数之间传递数据。

我们唯一需要做的是利用Python的sys.stdin读取输入数据,并把我们的输出传送给sys.stdout。Hadoop流将会帮助我们处理别的任何事情。

Map阶段:mapper.py

 #!/usr/bin/env python3
import sys
for line in sys.stdin:
line = line.strip()
words = line.split()
for word in words:
print("%s\t%s" % (word, 1))

Reducer阶段:reducer.py

 #!/usr/bin/env python3
from operator import itemgetter
import sys current_word = None
current_count = 0
word = None for line in sys.stdin:
line = line.strip()
word, count = line.split('\t', 1)
try:
count = int(count)
except ValueError: #count如果不是数字的话,直接忽略掉
continue
if current_word == word:
current_count += count
else:
if current_word:
print("%s\t%s" % (current_word, current_count))
current_count = count
current_word = word if word == current_word: #最后一个单词
print("%s\t%s" % (current_word, current_count))

python代码放在本地即可,不需上传到HDFS。由于后面需要执行这两段代码,所以为它们增加可执行权限,即:

chmod +x mapper.py
chmod +x reducer.py

本地测试

用Hadoop Streaming的好处之一就是因为代码没有库的依赖,调试方便,可以脱离Hadoop先在本地用管道模拟调试,所以我们先在本地进行测试。

mapper.py

Python初次实现MapReduce——WordCount

reducer.py

Python初次实现MapReduce——WordCount

Hadoop运行

数据准备

测试文件in.txt文件内容为:

Python初次实现MapReduce——WordCount

需要将其上传至HDFS,上传命令为:

bin/hadoop -copyFromLocal in.txt in.txt

Hadoop Streaming简介

Hadoop Streaming框架,最大的好处是,让任何语言编写的map, reduce程序能够在hadoop集群上运行,map/reduce程序只要遵循从标准输入stdin读,写出到标准输出stdout即可。

它通过将其他语言编写的 mapper 和 reducer 通过参数传给一个事先写好的 Java 程序(Hadoop 自带的 *-streaming.jar),这个 Java 程序会负责创建 MR 作业,另开一个进程来运行 mapper,将得到的输入通过 stdin 传给它,再将 mapper 处理后输出到 stdout 的数据交给 Hadoop,经过 partition 和 sort 之后,再另开进程运行 reducer,同样通过 stdin/stdout 得到最终结果。因此,我们只需要在其他语言编写的程序中,通过 stdin 接收数据,再将处理过的数据输出到 stdout,Hadoop Streaming 就能通过这个 Java 的 wrapper 帮我们解决中间繁琐的步骤,运行分布式程序。

Python初次实现MapReduce——WordCount

优点:

1. 可以使用自己喜欢的语言来编写 MapReduce 程序(不必非得使用 Java)

2. 不需要像写 Java 的 MR 程序那样 import 一大堆库,在代码里做很多配置,很多东西都抽象到了 stdio 上,代码量显著减少。

3. 因为没有库的依赖,调试方便,并且可以脱离 Hadoop 先在本地用管道模拟调试。

缺点:

1. 只能通过命令行参数来控制 MapReduce 框架,不像 Java 的程序那样可以在代码里使用 API,控制力比较弱。

2. 因为中间隔着一层处理,效率会比较慢。

3. 所以 Hadoop Streaming 比较适合做一些简单的任务,比如用 Python 写只有一两百行的脚本。如果项目比较复杂,或者需要进行比较细致的优化,使用 Streaming 就容易出现一些束手束脚的地方。

Hadoop Streaming运行

首先需要找到hadoop-streaming的位置,我的hadoop是2.x版本的,该包的位置在:

Python初次实现MapReduce——WordCount

在执行的过程中遇到了权限不够的问题:

Python初次实现MapReduce——WordCount

解决办法是扩大权限:

Python初次实现MapReduce——WordCount

为了方便起见,接下来我就把hadoop-streaming-2.9.2.jar放在了/usr/local/hadoop目录下,所以在下面的命令中大家注意一下。

最后输入如下命令:

bin/hadoop jar /usr/local/hadoop/hadoop-streaming-2.9.2.jar\
-mapper /usr/local/hadoop/mapper.py\
-file /usr/local/hadoop/mapper.py\
-reducer /usr/local/hadoop/reducer.py\
-file /usr/local/hadoop/reducer.py\
-input in.txt\
-output out

第一行是告诉Hadoop运行Streaming的Jav 程序,后面的mapper.py 和 reducer.py 是 mapper 所对应 Python 程序的路径。为了让Hadoop 将程序分发给其他机器,需要再加一个 -file 参数用于指明要分发的程序放在哪里。

Python初次实现MapReduce——WordCount

Python初次实现MapReduce——WordCount

Python代码优化

使用 Python 编写 Hadoop Streaming 时,在能使用 iterator 的情况下,尽量使用 iterator,避免将 stdin 的输入大量储存在内存里,否则会严重降低性能。

参考:

[1] 用python写MapReduce函数——以WordCount为例

[2] 使用Python实现Hadoop MapReduce程序

[3] Hadoop Streaming详解

[4] Hadoop Streaming 使用及参数设置

[5] 使用hadoop-streaming初体验mapreduce

[6] 使用python+hadoop-streaming编写hadoop处理程序