目前Hadoop是大家用的最普遍的大数据处理平台，但Hadoop本身是用JAVA开发的，作为PHP开发人员想要掺和大数据咋整呢？还好，Hadoop提供了大量的工具给其他语言的开发者。Hadoop Streaming就是Hadoop提供给其他语言开发者开发Mapreduce程序的工具。下边我们来体验一下PHP开发、测试和运行Mapreduce任务。
一、准备
- linux 操作系统
- PHP
- 搭建Hadoop环境（超出本文范围，请参考手册）
二、原理
Hadoop Streaming是Hadoop工具集中的一个工具，说白了就是一个jar包，它负责从标准输出读取数据，并将中间结果输出到标准输出，同时负责创建、提交和监控mapreduce任务。那么这就为我们提供了一个思路，我们只需要和从标准输出读取和输出数据一样写代码即可。
上述jar包一般在Hadoop的安装目录下，文件位置和文件名根据版本号不同而不同，比如2.6.2版如下：
$HAOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.6.2.jar
可以在Hadoop的安装目录下搜索文件名即可：
find -name '*streaming*'
Hadoop Streaming基本用法：
hadoop jar hadoop-streaming-2.6.2.jar \
-input myInputDirs \
-output myOutputDir \
-mapper /bin/cat \
-reducer /usr/bin/wc
此命令不赘述，后边实际用到我们再解释，如果想学习具体细节请查看参考手册
三、Mapreduce简要介绍
Mapreduce是Hadoop处理数据的框架。Mapreduce主要包括两个过程，一个是Map过程，在集群内的机器中分节点处理数据；一个是Reduce过程，将Map过程产生的结果进行处理。见下图(图片来源)
根据上述理解，我们就是要开发map程序和reduce程序，下边我们用PHP实现。
四、任务简述
接下来，我们实现如下功能：
使用PHP开发语言，利用Mapreduce处理用户数据，统计每个城市的注册用户数。
用户数据如下(users.txt)：
1,张三，23，beijing,10086, 2,李四，34，shanghai,10000, 3,王五，20，beijing,10010, ……
基本结构为“ID，name, age, city, telphone，”，每条数据一行。
五、Mapper实现
mapper.php
#!/usr/bin/php
<?php
$count = 0;
while($line = fgets(STDIN)) {
$line = trim($line);
$user = explode(',', $line);
echo $user[3]." 1\n";
}
这个脚本没啥特别的，其实就是CLI模式的PHP脚本，大家可能注意到了第一行不是<?php 而是#!/usr/bin/php。这个需要稍微说明一下，这个是所谓的shebang，通常我们执行PHP脚本，可能这么执行php mapper.php，如果加了第一行，我们只需要这样执行即可：./mapper.sh（是不是很眼熟！）。如果你不知道PHP安装在哪里，可以通过which php来找到它的路径。
此外因为Hadoop认为脚本都是可执行的程序，那么我们需要给这个mapper.php赋予可执行权限：
chmod +x mapper.php
上述程序很简单，只是从标准输出(STDIN)读取数据，然后取出城市信息，并标记一个1，每行一个输出出来。
六、Reducer实现
#!/usr/bin/php
<?php
$result = array();
while($line = fgets(STDIN)) {
list($city, $count) = explode(' ', $line);
if(!isset($result[$city])) $result[$city] = 0;
$result[$city] += $count;
}
foreach($result as $key=>$value){
echo "$key $value\n";
}
这段程序也比较简单，就是从标准输出(STDIN)按行读取数据，然后统计各城市的人数而已。同样我们需要shebang，并给reducer.php赋予可执行权限。
七、调试代码
目前我们得到了三个原材料文件:
- user.txt 存储用户数据信息，每行代表一条用户信息(可填充少量数据进行测试使用)。
- mapper.php 可执行的mapper脚本，具有shabang，并已赋予可执行权限。
- reducer.php 可执行的reducer脚本，具有shabang，并已赋予可执行权限。
调试mapper程序：
$ cat user.txt|./mapper.php shandong 1 beijing 1 beijing 1
将少量数据放入user.txt中，使用Linux命令行工具进行调试。
调试reducer程序：
$ cat user.txt | ./mapper.php | ./reducer.php beijing 2 shandong 1
实际上，reducer负责处理多个mapper产生的结果，此处我们仅模拟一个mapper输出然后用reducer处理即可，就能够完成任务目标，通过测试。
八、实际运行
8.1 启动Hadoop伪分布式集群
$HADOOP_HOME/sbin/start-dfs.sh $HADOOP_HOME/sbin/start-yar.sh
运行完上述两条命令，使用jps命令验证一下，如果有DataNode、 NodeManager、NameNode、ResourceManager、SecondaryNameNode说明启动成功。
8.2 创建目录并将文件存储到HDFS中
$HADOOP_HOME/bin/hdfs dfs -mkdir /user $HADOOP_HOME/bin/hdfs dfs -mkdi /user/<username> $HADOOP_HOME/bin/hdfs dfs -mkdi /user/<username>/input $HADOOP_HOME/bin/hdfs dfs -put user.txt /user/<username>/input
此处一定要注意<username>就是你当前登陆Linux系统的用用户名，比如我的是’hadoop’。最后一条命令一定要在user.txt所在目录执行，你懂的。
8.3 执行MAPpreduce命令
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.6.2.jar \ -input input/user.txt \ -output output2 \ -mapper /home/hadoop/Documents/src/php/mapper.php \ -reducer /home/hadoop/Documents/src/php/reducer.php
Hadoop可以运行jar文件，用java写代码就是用了这个原理，我们没辙只能通过streaming来搞，所以传入的是文章开头提到的那个hadoop-streaming文件。主要参数：
- input 输入的文件路径，这里是指HDFS中的路径。
- output Mapreduce运行结果输出目录，也是指HDFS中的目录，这个目录必须是不存在，Hadoop会自动重建，否则报路径已存在错误。
- mapper 我们的Mapper脚本，即mapper.php， 但必须写绝对路径，这里指本地文件路径，如果写相对路径或者脚本执行错误，会报配置文件配置错误。
- reducer 我们的Reducer脚本，即reducer.php, 同样需要绝对路径。
执行结束输出：
packageJobJar: [/tmp/hadoop-unjar2712470127417447142/] [] /tmp/streamjob561393005597605928.jar tmpDir=null
15/11/28 00:04:34 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 00:04:34 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 00:04:35 INFO mapred.FileInputFormat: Total input paths to process : 1
15/11/28 00:04:35 INFO mapreduce.JobSubmitter: number of splits:2
15/11/28 00:04:36 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1448639746810_0003
15/11/28 00:04:36 INFO impl.YarnClientImpl: Submitted application application_1448639746810_0003
15/11/28 00:04:36 INFO mapreduce.Job: The url to track the job: http://localhost:8088/proxy/application_1448639746810_0003/
15/11/28 00:04:36 INFO mapreduce.Job: Running job: job_1448639746810_0003
15/11/28 00:04:42 INFO mapreduce.Job: Job job_1448639746810_0003 running in uber mode : false
15/11/28 00:04:42 INFO mapreduce.Job: map 0% reduce 0%
15/11/28 00:04:51 INFO mapreduce.Job: map 100% reduce 0%
15/11/28 00:04:56 INFO mapreduce.Job: map 100% reduce 100%
15/11/28 00:04:57 INFO mapreduce.Job: Job job_1448639746810_0003 completed successfully
15/11/28 00:04:57 INFO mapreduce.Job: Counters: 49
File System Counters
FILE: Number of bytes read=46
FILE: Number of bytes written=324442
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=338
HDFS: Number of bytes written=23
HDFS: Number of read operations=9
HDFS: Number of large read operations=0
HDFS: Number of write operations=2
Job Counters
Launched map tasks=2
Launched reduce tasks=1
Data-local map tasks=2
Total time spent by all maps in occupied slots (ms)=12457
Total time spent by all reduces in occupied slots (ms)=3178
Total time spent by all map tasks (ms)=12457
Total time spent by all reduce tasks (ms)=3178
Total vcore-seconds taken by all map tasks=12457
Total vcore-seconds taken by all reduce tasks=3178
Total megabyte-seconds taken by all map tasks=12755968
Total megabyte-seconds taken by all reduce tasks=3254272
Map-Reduce Framework
Map input records=3
Map output records=3
Map output bytes=34
Map output materialized bytes=52
Input split bytes=200
Combine input records=0
Combine output records=0
Reduce input groups=2
Reduce shuffle bytes=52
Reduce input records=3
Reduce output records=2
Spilled Records=6
Shuffled Maps =2
Failed Shuffles=0
Merged Map outputs=2
GC time elapsed (ms)=668
CPU time spent (ms)=2320
Physical memory (bytes) snapshot=674447360
Virtual memory (bytes) snapshot=6298329088
Total committed heap usage (bytes)=491257856
Shuffle Errors
BAD_ID=0
CONNECTION=0
IO_ERROR=0
WRONG_LENGTH=0
WRONG_MAP=0
WRONG_REDUCE=0
File Input Format Counters
Bytes Read=138
File Output Format Counters
Bytes Written=23
15/11/28 00:04:57 INFO streaming.StreamJob: Output directory: output2
8.4 查看结果
$ $HADOOP_HOME/bin/hdfs dfs -cat output2/* beijing 2 shandong 1
由于运算结果是保存在HDFS中的，所以我们需要使用hdfs的cat命令去查看。
九、趟过的坑
如下遇到的问题，均为复现而得到。
9.1 文件夹已存在
packageJobJar: [/tmp/hadoop-unjar986578381948307717/] [] /tmp/streamjob9098881249462859146.jar tmpDir=null 15/11/28 23:46:00 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032 15/11/28 23:46:01 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032 15/11/28 23:46:01 ERROR streaming.StreamJob: Error Launching job : Output directory hdfs://localhost:9000/user/hadoop/output2 already exists Streaming Command Failed!
从错误的内容可以看出是目录已经存在，解决办法就是删除它：
$HADOOP_HOME/bin/hdfs dfs -rm -r -f output2
9.2 等待输出错误
packageJobJar: [/tmp/hadoop-unjar7696480168906810677/] [] /tmp/streamjob226646231265106849.jar tmpDir=null
15/11/28 23:52:02 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 23:52:02 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 23:52:03 INFO mapred.FileInputFormat: Total input paths to process : 1
15/11/28 23:52:03 INFO mapreduce.JobSubmitter: number of splits:2
15/11/28 23:52:04 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1448639746810_0004
15/11/28 23:52:04 INFO impl.YarnClientImpl: Submitted application application_1448639746810_0004
15/11/28 23:52:04 INFO mapreduce.Job: The url to track the job: http://localhost:8088/proxy/application_1448639746810_0004/
15/11/28 23:52:04 INFO mapreduce.Job: Running job: job_1448639746810_0004
15/11/28 23:52:12 INFO mapreduce.Job: Job job_1448639746810_0004 running in uber mode : false
15/11/28 23:52:12 INFO mapreduce.Job: map 0% reduce 0%
15/11/28 23:52:28 INFO mapreduce.Job: map 100% reduce 0%
15/11/28 23:52:31 INFO mapreduce.Job: Task Id : attempt_1448639746810_0004_m_000001_0, Status : FAILED
Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 2
at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:322)
at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:535)
at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:130)
at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:61)
at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:34)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:450)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:343)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1656)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
15/11/28 23:52:31 INFO mapreduce.Job: Task Id : attempt_1448639746810_0004_m_000000_0, Status : FAILED
Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 2
at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:322)
at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:535)
at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:130)
at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:61)
at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:34)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:450)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:343)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1656)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
这个错误是因为Hadoop无法执行你的程序，原因是你的脚本中没有添加shabang。
9.3 配置错误
packageJobJar: [/tmp/hadoop-unjar5507801207525106156/] [] /tmp/streamjob5584049820660820845.jar tmpDir=null
15/11/28 23:55:27 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 23:55:28 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
15/11/28 23:55:28 INFO mapred.FileInputFormat: Total input paths to process : 1
15/11/28 23:55:28 INFO mapreduce.JobSubmitter: number of splits:2
15/11/28 23:55:29 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1448639746810_0005
15/11/28 23:55:29 INFO impl.YarnClientImpl: Submitted application application_1448639746810_0005
15/11/28 23:55:30 INFO mapreduce.Job: The url to track the job: http://localhost:8088/proxy/application_1448639746810_0005/
15/11/28 23:55:30 INFO mapreduce.Job: Running job: job_1448639746810_0005
15/11/28 23:55:37 INFO mapreduce.Job: Job job_1448639746810_0005 running in uber mode : false
15/11/28 23:55:37 INFO mapreduce.Job: map 0% reduce 0%
15/11/28 23:55:45 INFO mapreduce.Job: Task Id : attempt_1448639746810_0005_m_000001_0, Status : FAILED
Error: java.lang.RuntimeException: Error in configuring object
at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:109)
at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:75)
at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:446)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:343)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1656)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
Caused by: java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:106)
... 9 more
Caused by: java.lang.RuntimeException: Error in configuring object
at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:109)
at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:75)
at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:133)
at org.apache.hadoop.mapred.MapRunner.configure(MapRunner.java:38)
... 14 more
Caused by: java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:106)
... 17 more
Caused by: java.lang.RuntimeException: configuration exception
at org.apache.hadoop.streaming.PipeMapRed.configure(PipeMapRed.java:222)
at org.apache.hadoop.streaming.PipeMapper.configure(PipeMapper.java:66)
... 22 more
Caused by: java.io.IOException: Cannot run program "/home/hadoop/Documents/src/php/mapper.php": error=13, Permission denied
at java.lang.ProcessBuilder.start(ProcessBuilder.java:1048)
at org.apache.hadoop.streaming.PipeMapRed.configure(PipeMapRed.java:209)
... 23 more
Caused by: java.io.IOException: error=13, Permission denied
at java.lang.UNIXProcess.forkAndExec(Native Method)
at java.lang.UNIXProcess.(UNIXProcess.java:248)
at java.lang.ProcessImpl.start(ProcessImpl.java:134)
at java.lang.ProcessBuilder.start(ProcessBuilder.java:1029)
... 24 more
这个错误比较隐晦，只提醒配置错误，起初以为是配置文件找不到，经过“苦心钻研”才发现，这里的配置就是指我们的input/output/mapper/reducer这几个参数的值，触发这个错误原因有很多，包括但不限于如下几种情况：
- mapper或者reducer程序有错误
- mapper或者reducer程序路径找不到，所以要写绝对路径
- mapper或者reducer程序没有赋予可执行权限，`chmod +x mapper.php`
十、总结
经过上述艰难的若干步，我们就完成了第一个Mapreduce程序的PHP版本，其实PHP在处理文本上具有先天性的优势，比如处理json或者XML等，感觉上PHP还是很适合开发Mapreduce程序的。
上述例子，我们完成了根据城市统计注册用户数，这个例子如果用SQL表示大约是这样：
SELCT city, count(*) _cnt FROM user GROUP BY city
本文仅用于展示使用PHP如何开发、测试和运行Mapreduce任务，关于上述程序仅用于参考。