您的位置:首页 > 编程语言 > Python开发

[python]使用python实现Hadoop MapReduce程序:计算一组数据的均值和方差

2016-03-17 12:43 1231 查看
这是参照《机器学习实战》中第15章“大数据与MapReduce”的内容,因为作者写作时hadoop版本和现在的版本相差很大,所以在Hadoop上运行python写的MapReduce程序时出现了很多问题,因此希望能够分享一些过程中的经验,但愿大家能够避开同样的坑。文章内容分为以下几个部分:(本文的代码和用到的数据集可以在这里下载)

1.代码分析

2.运行步骤

3.问题解决

1.代码分析

问题描述:在一个海量数据上分布式计算均值和方差的MapReduce作业。

设有一组数字,这组数字的均值和方差如下:



每个部分的{count(元素个数)、sum1/count、sum2/count},然后在reduce端将所有map端传入的sum1加起来在除以总个数n得到均值mean;将所有的sum2加起来除以n再减去均值mean的平方,就得到了方差var.

数据格式如下:一行包含一个数字,保存在inputFile.txt



Map端的代码如下:(保存在Mapper.py文件中)

#!/usr/bin/env python
#coding=utf-8
import sys
from numpy import mat, mean, power

def read_input(file):
for line in file:
yield line.rstrip()#rstrip()去除字符串右边的空格

input = read_input(sys.stdin)#依次读取每行的数据
input = [float(line) for line in input] #将每行转换成float型
numInputs = len(input)
input = mat(input)
sqInput = power(input,2)

#输出数据个数,均值,以及平方和的均值,以'\t'隔开
print "%d\t%f\t%f" % (numInputs, mean(input), mean(sqInput))


这里补充说明一下,在read_input()函数中,为何要使用yield?这里使用的是海量数据集,如果直接对文件对象调用 read() 方法,会导致不可预测的内存占用。好的方法是利用固定长度的缓冲区来不断读取文件内容。通过 yield,我们不再需要编写读文件的迭代类,就可以轻松实现文件读取。

下面是Reduce端的代码(保存在Reducer.py文件中),它接收map端的输出,并将数据合并成全局的均值,并计算得到方差。

#!/usr/bin/env python
#coding=utf-8

import sys
from numpy import mat, mean, power

def read_input(file):
for line in file:
yield line.rstrip()

input = read_input(sys.stdin)

#读取map端的输出,共有三个字段,按照'\t'分隔开来
mapperOut = [line.split('\t') for line in input]

cumVal=0.0
cumSumSq=0.0
cumN=0.0
for instance in mapperOut:
nj = float(instance[0])#第一个字段是数据个数
cumN += nj
cumVal += nj*float(instance[1])#第二个字段是一个map输出的均值,均值乘以数据个数就是数据总和
cumSumSq += nj*float(instance[2])#第三个字段是一个map输出的平方和的均值,乘以元素个数就是所有元素的平方和

mean = cumVal/cumN#得到所有元素的均值
var = (cumSumSq/cumN-mean*mean)#得到所有元素的方差

print "%d\t%f\t%f" % (cumN, mean, var)


2.运行步骤

我使用的环境是:

Centos 64

Python 2.6

Hadoop 2.2.0

2.1 本地运行

在运行之前,首先在本地运行一下,看是否能通过。

首先将以上Mapper.py和Reducer.py文件,以及数据文件inputFile.txt放在同一个文件夹中(我这里是桌面的文件夹:/home/hadoop/Desktop/python_doc中),然后输入命令:chmod +x文件名,修改其权限变成可执行文件。

然后输入以下命令:

[hadoop@hadoop1 python_doc]$ python Mapper.py < inputFile.txt | python Reducer.py




出现上述结果表示运行通过。

2.2 hadoop上运行

1.启动HDFS,进入HADOOP_HOME目录(也就是hadoop的安装目录,我的是/app/hadoop/hadoop):

[hadoop@hadoop1 python_doc]$cd $HADOOP_HOME/sbin
[hadoop@hadoop1 sbin]$./start-dfs.sh


2.验证HDFS是否启动,在MasterNode上输入以下命令,将会出现:NameNode、SecondaryNameNode和DataNode

[hadoop@hadoop1 sbin]$./jps




3.在HDFS下创建一个存放“输入数据“的文件夹

[hadoop@hadoop1 Desktop]$ hadoop fs -mkdir /user/hadoop/mr-input


注意:这里不需要创建“输出数据“的文件夹,否则会出现以下错误:

ERROR security.UserGroupInformation: PriviledgedActionException as:hadoop (auth:SIMPLE) cause:org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://hadoop1:9000/user/hadoop/mr-ouput15 already exists


4.将数据文件inputFile.txt复制到HDFS:

[hadoop@hadoop1 python_doc]$ hadoop fs -put inputFile.txt  /user/hadoop/mr-input


也可以查看一下,文件是否复制成功:

[hadoop@hadoop1 python_doc]$ hadoop fs –ls  /user/hadoop/mr-input


5.下面重点来了,在命令窗口输入Hadoop Streaming命令:

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-*streaming*.jar \
-input /user/hadoop/mr-input/* \
-output /user/hadoop/mr-ouput13 \
-file /home/hadoop/Desktop/python_doc/Mapper.py -mapper 'Mapper.py' \
-file /home/hadoop/Desktop/python_doc/Reducer.py -reducer 'Reducer.py'


注意:在hadoop2.2.0的版本中,streaming.jar的目录发生了改变,保存在:$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar

这里再补充一下,Hadoop Streaming的用法:

Hadoop Streaming用法
Usage: $HADOOP_HOME/bin/hadoop jar \
$HADOOP_HOME/contrib/streaming/hadoop-*streaming*.jar [options]
options:
(1)-input:输入文件路径
(2)-output:输出文件路径
(3)-mapper:用户自己写的mapper程序
(4)-reducer:用户自己写的reducer程序
(5)-file:打包文件到提交的作业中,可以是mapper或者reducer要用的输入文件。
(6)-partitioner:用户自定义的partitioner程序
(7)-combiner:用户自定义的combiner程序(必须用java实现)
(8)-D:作业的一些属性(以前用的是-jonconf),具体有:
1)mapred.map.tasks:map task数目
2)mapred.reduce.tasks:reduce task数目
3)stream.map.input.field.separator/stream.map.output.field.separator: map task输入/输出数据的分隔符,默认均为\t。
4)stream.num.map.output.key.fields:指定map task输出记录中key所占的域数目
5)stream.reduce.input.field.separator/stream.reduce.output.field.separator:reduce task输入/输出数据的分隔符,默认均为\t。
6)stream.num.reduce.output.key.fields:指定reduce task输出记录中key所占的域数目


6. 运行完成后:

查看在输出文件夹下的内容:

[hadoop@hadoop1 Desktop]$ hadoop fs -ls /user/hadoop/mr-ouput13
Found 2 items
-rw-r--r--   2 hadoop supergroup          0 2016-03-16 16:45 /user/hadoop/mr-ouput13/_SUCCESS
-rw-r--r--   2 hadoop supergroup         22 2016-03-16 16:45 /user/hadoop/mr-ouput13/part-00000


查看结果,结果是输出文件夹中的part-00000文件(显示计算结果与本地计算结果是一致的)

[hadoop@hadoop1 Desktop]$ hadoop fs -cat  /user/hadoop/mr-ouput13/part-00000


3.问题解决

1. 出现“Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1”的错误

解决方法:

确保Mapper.py Reducer.py这两个文件的权限是可执行的,如果不是可执行的,使用:chmod +x 文件名,修改其权限为可执行的。

确保安装numpy包,Centos下的安装方法是:

sudo yum -y install gcc gcc-c++ numpy python-devel scipy


这个命令会自动把依赖的包都装好。安装完成后,测试一下:

[hadoop@hadoop1 python_doc]$ python
Python 2.6.6 (r266:84292, Jul 23 2015, 15:22:56)
[GCC 4.4.7 20120313 (Red Hat 4.4.7-11)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> from numpy import mat


没有提示错误,说明numpy包已安装好。

在hadoop上实施MapReduce之前,一定要在本地运行一下你的python程序,看是否能够跑通。

首先进入包含map和reduce两个py脚本文件和数据文件inputFile.txt的文件夹中。然后输入一下命令,看是否执行通过:

[hadoop@hadoop1 python_doc]$ python Mapper.py < inputFile.txt | python Reducer.py


2.出现错误:“Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 2”,或者出现jar文件找不到的情况,或者出现输出文件夹已经存在的情况。

Mapper.py和Reduce.py的最前面要加上:#!/usr/bin/env python 这条语句

在Hadoop Streaming命令中,请确保按以下的格式来输入

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-*streaming*.jar \
-input /user/hadoop/mr-input/* \
-output /user/hadoop/mr-ouput13 \
-file /home/hadoop/Desktop/python_doc/Mapper.py -mapper 'Mapper.py' \
-file /home/hadoop/Desktop/python_doc/Reducer.py -reducer 'Reducer.py'


要确保jar文件的路径正确,hadoop 2.2版本的该文件是保存在:$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar中,不同的hadoop版本可能略有不同;

保存数据文件的HDFS文件夹后要加上”/*”,我这里是“/user/hadoop/mr-input”目录,加上”/*”之后表示该文件夹下所有的文件作为输入的数据文件;

HDFS中的输出文件夹(这里是HDFS下的/user/hadoop/mr-ouput13),一定要是一个新的(之前不存在)的文件夹,因为即使上条Hadoop Streaming命令没有执行成功,仍然会根据你的命令来创建输出文件夹,而后面再输入Hadoop Streaming命令如果使用相同的输出文件夹时,就会出现“输出文件夹已经存在的错误”;

参数 –file后面是map和reduce的脚本,路径是详细的绝对路径(我这里是/home/hadoop/Desktop/python_doc/Mapper.py),但是在参数 -mapper 和-reducer之后,文件名只需要python脚本的名字即可,而且用引号引起来(比如我这里是:-mapper 'Mapper.py')

Reference:

Peter Harrington,《机器学习实战》,人民邮电出版社,2013

http://stackoverflow.com/questions/4460522/hadoop-streaming-job-failed-error-in-python (Stackoverflow上关于Hadoop Streaming命令失败的解答)

http://dongxicheng.org/mapreduce/hadoop-streaming-programming/ (Hadoop Streaming 的参数介绍)
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: