用python + hadoop streaming 编写分布式程序（二）

qingkuangs

写在前面
　　相关随笔：

• Hadoop-1.0.4集群搭建笔记
  
• 用python + hadoop streaming 编写分布式程序（一） -- 原理介绍，样例程序与本地调试
  
• 用python + hadoop streaming 编写分布式程序（三） -- 自定义功能
  

　　
　　为了方便，这篇文章里的例子均为伪分布式运行，一般来说只要集群配置得当，在伪分布式下能够运行的程序，在真实集群上也不会有什么问题。
　　为了更好地模拟集群环境，我们可以在mapred-site.xml中增设reducer和mapper的最大数目（默认为2，实际可用数目大约是CPU核数-1）。
　　假设你为Hadoop安装路径添加的环境变量叫$HADOOP_HOME（如果是$HADOOP_PREFIX，下文看到的命令对应改改就行）



$ vi $HADOOP_HOME/conf/mapred-site.xml

　　假设这台机子的CPU是4核，那么可以添加下面这几行




mapred.tasktracker.reduce.tasks.maximum
3


mapred.tasktracker.map.tasks.maximum
3


　　这里修改了以后只是增加了slot的数量，如果你在执行MR作业时没有指明需要多少mapper或reducer，不一定会用到这么多，Hadoop会自行分配。

查看文档
　　首先需要知道用于streaming的java程序在哪里。在1.0.x的版本中，应该都在$HADOOP_HOME/contrib/streaming/下。比如1.0.4的就在



$HADOOP_HOME/contrib/streaming/hadoop-streaming-1.0.4.jar

　　里。 首先我们可以先看看这个java程序自带的文档。以下以1.0.4版本为例，执行



$ hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-streaming-1.0.4.jar -info

　　就会看到一系列自带的帮助，带有各种参数的说明和一些使用样例。

运行命令
　　用Hadoop Streaming执行python程序的一般步骤是：

• 　　将输入文件放到HDFS上，建议使用copyFromLocal而不是put命令，参见Difference between hadoop fs -put and hadoop fs -copyFromLocal
  
  
  
  • 　　一般可以新建一个文件夹用于存放输入文件，假设叫input
    
    
    
    $ hadoop fs -mkdir input
    
    　　然后用
    
    
    
    $ hadoop fs -ls
    
    　　查看目录，可以看到出现了一个/user/hadoop/input文件夹。/user/hadoop是默认的用户文件夹，相当于本地文件系统中的/home/hadoop。
    
    
  • 　　再使用
    
    
    
    $ hadoop fs -copyFromLocal  input/
    
    　　将本地文件放到input文件夹下。copyFromLocal命令的用法类似于Linux的cp命令，支持使用wildcard。如果出现了预期外的执行结果，可以试试看在使用了wildcard的路径外加上引号，参见官方FAQ
    
    
  　　建议阅读：HDFS命令文档
  
  
  
• 　　开始MR作业，以1.0.4版本为例，假设你现在正在放有mapper和reducer两个脚本的目录下，而且它们刚好就叫mapper.py和reducer.py，在不需要做其他配置的情况下，执行
  
  
  
  $hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-streaming-1.0.4.jar \
  -mapper mapper.py   -file mapper.py\
  -reducer reducer.py  -file reducer.py \
  -input input/* -output output
  
  　　第一行是告诉Hadoop运行streaming的java程序，接下来的是参数：
  
  
  
  • 这里的mapper.py和reducer.py是mapper所在python程序的路径。为了让Hadoop将程序分发给其他机器，需要再加一个-file参数用于指明要分发的程序放在哪里。
    
  • 　　注意这样写的前提是这个python程序里有shebang而且添加了执行权限。如果没有的话，可以改成
    
    
    
    -mapper 'python mapper.py'
    
    　　加上解释器命令，用引号括住。因为准确来说，mapper后面跟的其实应该是一个命令而不是一个文件名。
    
    
  • 　　假如你执行的程序不放在当前目录下，比如说在当前目录的src文件夹下，可以这样写
    
    
    
    -mapper mapper.py   -file src/mapper.py\
    -reducer reducer.py  -file src/reducer.py \
    
    　　也就是说，-mapper和-reducer后面跟的文件名不需要带上路径，而-file后的参数则需要。注意如果你在mapper后的命令用了引号，加上路径名反而会报错说找不到这个程序。
    
    
  • -input和-output后面跟的是HDFS上的路径名，同样支持wildcard，这里的input/*指的就是“input文件夹下的所有文件”。注意-output后面跟着的需要是一个不存在于HDFS上的路径，在产生输出的时候hadoop会帮你创建这个文件夹，如果已经存在的话就会产生冲突。
    
  • 　　有时候shebang不一定能用，尤其是在执行环境比较复杂的时候。最保险的写法可能是：
    
    
    
    $hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-streaming-1.0.4.jar \
    -mapper 'python mapper.py'   -file mapper.py\
    -reducer 'python reducer.py'  -file reducer.py \
    -input input/* -output output
    
    　　这样写还有一个好处，就是可以在引号里写上提供给python程序的命令行参数，以后的教程会提到怎么用。
    
    
  • 由于mapper和reducer参数跟的实际上是命令，所以如果每台机器上python的环境配置不一样的话，会用每台机器自己的配置去执行python程序。
    
  
  
  

运行过程
　　写完命令回车，顺利的话会开始执行程序。 这里不赘述执行时输出到终端的内容，可以去这里看看正常运行的时候会给些什么。

利用WebUI监控集群运行情况
　　一般来说要检查运行状况，都是去jobtracker的webUI。如果在master上，用浏览器访问http://localhost:50030即可（如果你在配置hadoop的时候修改了mapred-site.xml的mapred.job.tracker.http.address，请访问对应的其他地址）
　　在webUI里你可以看到running jobs, completed jobs和retired jobs。点击Jobid下的超链接，可以看到对应job的执行状况。进去后如果看到Failed/Killed Task Attempts下非空，你可以点进对应的超链接，找到对应的log去进行debug。

得到结果
　　成功执行完这个任务之后，你用output参数在HDFS上指定的输出文件夹里就会多出几个文件

• 一个空白文件_SUCCESS，表明job运行成功，这个文件可以让其他程序只要查看一下HDFS就能判断这次job是否成功运行，从而进行相关处理。
  
• 一个_logs文件夹，顾名思义里面放着任务日志
  
• part-00000, .... part-xxxxx文件，有多少个reducer后面的数字就会有多大，对应每个reducer的输出结果。
  

　　假如你的输出很少，比如是一个只有几行的计数，你可以用



$ hadoop fs -cat

　　直接将输出打印到终端查看。
　　假如你的输出很多，则需要拷贝到本地文件系统来查看。可以使用copyToLocal来获取整个文件夹（与copyFromLocal一样，它与get的区别在于会限制目标文件夹一定在本地文件系统上）。如果你不需要_SUCCESS 和_logs，并且想要将所有reducer的输出合并，可以使用getmerge命令。
　　比如在上面的例子里，可以用命令



$ hadoop fs -copyToLocal output ./

　　将output文件夹复制到本地文件系统的当前目录下，或者用



$ hadoop fs -getmerge output ./

　　将output下的part-xxxxx合并，放到当前目录的一个叫output的文件里。

如何串联多趟MR
　　如果你有多次任务要执行，下一步需要用上一步的任务做输入，解决办法其实很简单。假设上一步在HDFS的输出文件夹是output1，那么在下一步的运行命令中，指明



-input output1/part-*

　　即指定上一次的所有输出为本次任务的输入即可。注意这里假设你不需要对上一步的输出做额外处理。

其他
　　这篇文章只提到了最简单的执行Hadoop streaming程序的方法。涉及到一些其他需求，比如需要有多个输入文件等情况，还需要进一步调整运行命令，会在以后的文章里讲到。

推荐阅读

• Writing an Hadoop MapReduce Program in Python
  
• Python Map Reduce on Hadoop - A Beginners Tutorial
  
• Writing Hadoop Applications in Python with Hadoop Streaming