Python初次实现MapReduce—

Python初次实现MapReduce——WordCount

news2024/11/27 0:31:53

前言

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

reducer.py

Hadoop运行

数据准备

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

需要将其上传至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 帮我们解决中间繁琐的步骤，运行分布式程序。