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

时间:2021-11-24 00:37:07

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

1.代码分析

2.运行步骤

3.问题解决

1.代码分析

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

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

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

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

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

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

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

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

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

 

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

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

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'
    1. 要确保jar文件的路径正确,hadoop 2.2版本的该文件是保存在:$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar,不同的hadoop版本可能略有不同;
    2. 保存数据文件的HDFS文件夹后要加上”/*”,我这里是“/user/hadoop/mr-input”目录,加上”/*”之后表示该文件夹下所有的文件作为输入的数据文件;
    3. HDFS中的输出文件夹(这里是HDFS下的/user/hadoop/mr-ouput13),一定要是一个新的(之前不存在)的文件夹,因为即使上条Hadoop Streaming命令没有执行成功,仍然会根据你的命令来创建输出文件夹,而后面再输入Hadoop Streaming命令如果使用相同的输出文件夹时,就会出现“输出文件夹已经存在的错误”;
    4. 参数 –file后面是map和reduce的脚本,路径是详细的绝对路径(我这里是/home/hadoop/Desktop/python_doc/Mapper.py),但是在参数 -mapper 和-reducer之后,文件名只需要python脚本的名字即可,而且用引号引起来(比如我这里是:-mapper 'Mapper.py'

Reference:

  1. Peter Harrington,《机器学习实战》,人民邮电出版社,2013
  2. http://*.com/questions/4460522/hadoop-streaming-job-failed-error-in-python (*上关于Hadoop Streaming命令失败的解答)
  3. http://dongxicheng.org/mapreduce/hadoop-streaming-programming/ (Hadoop Streaming 的参数介绍)