这是参照《机器学习实战》中第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) #输出数据个数，均值，以及平方和的均值，以' '隔开 print "%d %f %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端的输出，共有三个字段，按照' '分隔开来 mapperOut = [line.split(' ') 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 %f %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输入/输出数据的分隔符,默认均为 。 4）stream.num.map.output.key.fields：指定map task输出记录中key所占的域数目 5）stream.reduce.input.field.separator/stream.reduce.output.field.separator：reduce task输入/输出数据的分隔符，默认均为 。 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 的参数介绍)