简单之美 | Hadoop Streaming原理及实践

Hadoop Streaming提供了一个便于进行MapReduce编程的工具包，使用它可以基于一些可执行命令、脚本语言或其他编程语言来实现Mapper和 Reducer，从而充分利用Hadoop并行计算框架的优势和能力，来处理大数据。需要注意的是，Streaming方式是基于Unix系统的标准输入输出来进行MapReduce Job的运行，它区别与Pipes的地方主要是通信协议，Pipes使用的是Socket通信，是对使用C++语言来实现MapReduce Job并通过Socket通信来与Hadopp平台通信，完成Job的执行。任何支持标准输入输出特性的编程语言都可以使用Streaming方式来实现MapReduce Job，基本原理就是输入从Unix系统标准输入，输出使用Unix系统的标准输出。
Hadoop是使用Java语言编写的，所以最直接的方式的就是使用Java语言来实现Mapper和Reducer，然后配置MapReduce Job，提交到集群计算环境来完成计算。但是很多开发者可能对Java并不熟悉，而是对一些具有脚本特性的语言，如C++、Shell、Python、 Ruby、PHP、Perl有实际开发经验，Hadoop Streaming为这一类开发者提供了使用Hadoop集群来进行处理数据的工具，即工具包hadoop-streaming-.jar。
Hadoop Streaming使用了Unix的标准输入输出作为Hadoop和其他编程语言的开发接口，因此在其他的编程语言所写的程序中，只需要将标准输入作为程序的输入，将标准输出作为程序的输出就可以了。在标准的输入输出中，Key和Value是以Tab作为分隔符，并且在Reducer的标准输入中，Hadoop框架保证了输入的数据是经过了按Key排序的。
如何使用Hadoop Streaming工具呢？我们可以查看该工具的使用方法，通过命令行来获取，如下所示：

xiaoxiang@ubuntu3:~/hadoop$ bin/hadoop jar ./contrib/streaming/hadoop-streaming-1.0.3.jar -info
Usage: $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar [options]
Options:
  -input    <path>     DFS input file(s) for the Map step
  -output   <path>     DFS output directory for the Reduce step
  -mapper   <cmd|JavaClassName>      The streaming command to run
  -combiner <cmd|JavaClassName> The streaming command to run
  -reducer  <cmd|JavaClassName>      The streaming command to run
  -file     <file>     File/dir to be shipped in the Job jar file
  -inputformat TextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassName Optional.
  -outputformat TextOutputFormat(default)|JavaClassName  Optional.
  -partitioner JavaClassName  Optional.
  -numReduceTasks <num>  Optional.
  -inputreader <spec>  Optional.
  -cmdenv   <n>=<v>    Optional. Pass env.var to streaming commands
  -mapdebug <path>  Optional. To run this script when a map task fails
  -reducedebug <path>  Optional. To run this script when a reduce task fails
  -io <identifier>  Optional.
  -verbose

Generic options supported are
-conf <configuration file>     specify an application configuration file
-D <property=value>            use value for given property
-fs <local|namenode:port>      specify a namenode
-jt <local|jobtracker:port>    specify a job tracker
-files <comma separated list of files>    specify comma separated files to be copied to the map reduce cluster
-libjars <comma separated list of jars>    specify comma separated jar files to include in the classpath.
-archives <comma separated list of archives>    specify comma separated archives to be unarchived on the compute machines.

The general command line syntax is
bin/hadoop command [genericOptions] [commandOptions]


In -input: globbing on <path> is supported and can have multiple -input
Default Map input format: a line is a record in UTF-8
  the key part ends at first TAB, the rest of the line is the value
Custom input format: -inputformat package.MyInputFormat
Map output format, reduce input/output format:
  Format defined by what the mapper command outputs. Line-oriented

The files named in the -file argument[s] end up in the
  working directory when the mapper and reducer are run.
  The location of this working directory is unspecified.

To set the number of reduce tasks (num. of output files):
  -D mapred.reduce.tasks=10
To skip the sort/combine/shuffle/sort/reduce step:
  Use -numReduceTasks 0
  A Task's Map output then becomes a 'side-effect output' rather than a reduce input
  This speeds up processing, This also feels more like "in-place" processing
  because the input filename and the map input order are preserved
  This equivalent -reducer NONE

To speed up the last maps:
  -D mapred.map.tasks.speculative.execution=true
To speed up the last reduces:
  -D mapred.reduce.tasks.speculative.execution=true
To name the job (appears in the JobTracker Web UI):
  -D mapred.job.name='My Job'
To change the local temp directory:
  -D dfs.data.dir=/tmp/dfs
  -D stream.tmpdir=/tmp/streaming
Additional local temp directories with -cluster local:
  -D mapred.local.dir=/tmp/local
  -D mapred.system.dir=/tmp/system
  -D mapred.temp.dir=/tmp/temp
To treat tasks with non-zero exit status as SUCCEDED:
  -D stream.non.zero.exit.is.failure=false
Use a custom hadoopStreaming build along a standard hadoop install:
  $HADOOP_HOME/bin/hadoop jar /path/my-hadoop-streaming.jar [...]\
    [...] -D stream.shipped.hadoopstreaming=/path/my-hadoop-streaming.jar
For more details about jobconf parameters see:

http://wiki.apache.org/hadoop/JobConfFile

To set an environement variable in a streaming command:
   -cmdenv EXAMPLE_DIR=/home/example/dictionaries/

Shortcut:
   setenv HSTREAMING "$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar"

Example: $HSTREAMING -mapper "/usr/local/bin/perl5 filter.pl"
           -file /local/filter.pl -input "/logs/0604*/*" [...]
  Ships a script, invokes the non-shipped perl interpreter
  Shipped files go to the working directory so filter.pl is found by perl
  Input files are all the daily logs for days in month 2006-04

面，我们分别选择几个可以使用Hadoop Streaming工具来进行计算的例子，比如对单词词频进行统计计算，即WordCount功能。
首先，我们准备测试使用的数据集，如下所示：

xiaoxiang@ubuntu3:~/hadoop$ bin/hadoop fs -lsr /user/xiaoxiang/dataset/join/
-rw-r--r--   3 xiaoxiang supergroup  391103029 2013-03-26 12:19 /user/xiaoxiang/dataset/join/irc_basic_info.ds
-rw-r--r--   3 xiaoxiang supergroup   11577164 2013-03-26 12:20 /user/xiaoxiang/dataset/join/irc_net_block.ds
-rw-r--r--   3 xiaoxiang supergroup    8335235 2013-03-26 12:20 /user/xiaoxiang/dataset/join/irc_org_info.ds

一共有3个数据文件，大概将近400M大小。
下面，选择Python语言来实现MapReduce Job的运行。
使用Python实现Mapper，代码文件为word_count_mapper.py，代码如下所示：

#!/usr/bin/env python

import sys

for line in sys.stdin:
    line = line.strip()
    words = filter(lambda word: word, line.split())
    for word in words:
        print '%s\t%s' % (word, 1)

使用Python实现Reducer，代码文件为word_count_reducer.py，代码如下所示：

#!/usr/bin/env python

import sys
from operator import itemgetter

wc_dict = {}

for line in sys.stdin:
    line = line.strip()
    word, count = line.split()
    try:
        count = int(count)
        wc_dict[word] = wc_dict.get(word, 0) + count
    except ValueError:
        pass

sorted_dict = sorted(wc_dict.items(), key=itemgetter(0))
for word, count in sorted_dict:
    print '%s\t%s' % (word, count)

运行Python实现的MapReduce程序，如下所示：

xiaoxiang@ubuntu3:~/hadoop$ bin/hadoop jar ./contrib/streaming/hadoop-streaming-1.0.3.jar -input /user/xiaoxiang/dataset/join/ -output /user/xiaoxiang/output/streaming/python -mapper word_count_mapper.py -reducer word_count_reducer.py -numReduceTasks 2 -file *.py
packageJobJar: [word_count_mapper.py, word_count_reducer.py, /opt/stone/cloud/storage/tmp/hadoop-xiaoxiang/hadoop-unjar4066863202997744310/] [] /tmp/streamjob2336302975421423718.jar tmpDir=null
13/04/18 17:50:17 INFO util.NativeCodeLoader: Loaded the native-hadoop library
13/04/18 17:50:17 WARN snappy.LoadSnappy: Snappy native library not loaded
13/04/18 17:50:17 INFO mapred.FileInputFormat: Total input paths to process : 3
13/04/18 17:50:17 INFO streaming.StreamJob: getLocalDirs(): [/opt/stone/cloud/storage/mapred/local]
13/04/18 17:50:17 INFO streaming.StreamJob: Running job: job_201303302227_0047
13/04/18 17:50:17 INFO streaming.StreamJob: To kill this job, run:
13/04/18 17:50:17 INFO streaming.StreamJob: /opt/stone/cloud/hadoop-1.0.3/libexec/../bin/hadoop job  -Dmapred.job.tracker=hdfs://ubuntu3:9001/ -kill job_201303302227_0047
13/04/18 17:50:17 INFO streaming.StreamJob: Tracking URL: http://ubuntu3:50030/jobdetails.jsp?jobid=job_201303302227_0047
13/04/18 17:50:18 INFO streaming.StreamJob:  map 0%  reduce 0%
13/04/18 17:50:33 INFO streaming.StreamJob:  map 7%  reduce 0%
13/04/18 17:50:36 INFO streaming.StreamJob:  map 11%  reduce 0%
13/04/18 17:50:39 INFO streaming.StreamJob:  map 15%  reduce 0%
13/04/18 17:50:42 INFO streaming.StreamJob:  map 19%  reduce 0%
13/04/18 17:50:45 INFO streaming.StreamJob:  map 23%  reduce 0%
13/04/18 17:50:48 INFO streaming.StreamJob:  map 25%  reduce 0%
13/04/18 17:51:09 INFO streaming.StreamJob:  map 32%  reduce 2%
13/04/18 17:51:12 INFO streaming.StreamJob:  map 36%  reduce 4%
13/04/18 17:51:15 INFO streaming.StreamJob:  map 40%  reduce 8%
13/04/18 17:51:18 INFO streaming.StreamJob:  map 44%  reduce 8%
13/04/18 17:51:21 INFO streaming.StreamJob:  map 47%  reduce 8%
13/04/18 17:51:24 INFO streaming.StreamJob:  map 50%  reduce 8%
13/04/18 17:51:45 INFO streaming.StreamJob:  map 54%  reduce 10%
13/04/18 17:51:48 INFO streaming.StreamJob:  map 60%  reduce 15%
13/04/18 17:51:51 INFO streaming.StreamJob:  map 65%  reduce 17%
13/04/18 17:51:55 INFO streaming.StreamJob:  map 66%  reduce 17%
13/04/18 17:51:58 INFO streaming.StreamJob:  map 68%  reduce 17%
13/04/18 17:52:01 INFO streaming.StreamJob:  map 72%  reduce 17%
13/04/18 17:52:04 INFO streaming.StreamJob:  map 75%  reduce 17%
13/04/18 17:52:19 INFO streaming.StreamJob:  map 75%  reduce 19%
13/04/18 17:52:22 INFO streaming.StreamJob:  map 87%  reduce 21%
13/04/18 17:52:25 INFO streaming.StreamJob:  map 100%  reduce 23%
13/04/18 17:52:28 INFO streaming.StreamJob:  map 100%  reduce 27%
13/04/18 17:52:31 INFO streaming.StreamJob:  map 100%  reduce 29%
13/04/18 17:52:37 INFO streaming.StreamJob:  map 100%  reduce 49%
13/04/18 17:52:40 INFO streaming.StreamJob:  map 100%  reduce 69%
13/04/18 17:52:43 INFO streaming.StreamJob:  map 100%  reduce 72%
13/04/18 17:52:46 INFO streaming.StreamJob:  map 100%  reduce 74%
13/04/18 17:52:49 INFO streaming.StreamJob:  map 100%  reduce 76%
13/04/18 17:52:52 INFO streaming.StreamJob:  map 100%  reduce 78%
13/04/18 17:52:55 INFO streaming.StreamJob:  map 100%  reduce 79%
13/04/18 17:52:58 INFO streaming.StreamJob:  map 100%  reduce 81%
13/04/18 17:53:01 INFO streaming.StreamJob:  map 100%  reduce 84%
13/04/18 17:53:04 INFO streaming.StreamJob:  map 100%  reduce 87%
13/04/18 17:53:07 INFO streaming.StreamJob:  map 100%  reduce 90%
13/04/18 17:53:10 INFO streaming.StreamJob:  map 100%  reduce 93%
13/04/18 17:53:13 INFO streaming.StreamJob:  map 100%  reduce 96%
13/04/18 17:53:16 INFO streaming.StreamJob:  map 100%  reduce 98%
13/04/18 17:53:19 INFO streaming.StreamJob:  map 100%  reduce 99%
13/04/18 17:53:22 INFO streaming.StreamJob:  map 100%  reduce 100%
13/04/18 17:54:10 INFO streaming.StreamJob: Job complete: job_201303302227_0047
13/04/18 17:54:10 INFO streaming.StreamJob: Output: /user/xiaoxiang/output/streaming/python

验证结果，如下所示：

xiaoxiang@ubuntu3:~/hadoop$ bin/hadoop fs -lsr /user/xiaoxiang/output/streaming/python
-rw-r--r--   3 xiaoxiang supergroup          0 2013-04-18 17:54 /user/xiaoxiang/output/streaming/python/_SUCCESS
drwxr-xr-x   - xiaoxiang supergroup          0 2013-04-18 17:50 /user/xiaoxiang/output/streaming/python/_logs
drwxr-xr-x   - xiaoxiang supergroup          0 2013-04-18 17:50 /user/xiaoxiang/output/streaming/python/_logs/history
-rw-r--r--   3 xiaoxiang supergroup      37646 2013-04-18 17:50 /user/xiaoxiang/output/streaming/python/_logs/history/job_201303302227_0047_1366278617842_xiaoxiang_streamjob2336302975421423718.jar
-rw-r--r--   3 xiaoxiang supergroup      21656 2013-04-18 17:50 /user/xiaoxiang/output/streaming/python/_logs/history/job_201303302227_0047_conf.xml
-rw-r--r--   3 xiaoxiang supergroup   91367389 2013-04-18 17:52 /user/xiaoxiang/output/streaming/python/part-00000
-rw-r--r--   3 xiaoxiang supergroup   91268074 2013-04-18 17:52 /user/xiaoxiang/output/streaming/python/part-00001
xiaoxiang@ubuntu3:~/hadoop$ bin/hadoop fs -cat /user/xiaoxiang/output/streaming/python/part-00000 | head -5
!    2
#    36
#039    1
#1059)    1
#1098    1

相关问题

在使用Python实现MapReduce时，总是执行失败？
可以查看TaskTracker结点运行日志，可以看到，总是找不到对应的Python脚本文件，错误示例如下：

xiaoxiang@ubuntu1:/opt/stone/cloud/storage/logs/hadoop/userlogs/job_201303302227_0045/attempt_201303302227_0045_m_000001_0$ cat stderr
java.io.IOException: Cannot run program "/user/xiaoxiang/streaming/python/word_count_mapper.py": java.io.IOException: error=2, No such file or directory
    at java.lang.ProcessBuilder.start(ProcessBuilder.java:460)
    at org.apache.hadoop.streaming.PipeMapRed.configure(PipeMapRed.java:214)
    at org.apache.hadoop.streaming.PipeMapper.configure(PipeMapper.java:66)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:88)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:64)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:117)
    at org.apache.hadoop.mapred.MapRunner.configure(MapRunner.java:34)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.ReflectionUtils.setJobConf(ReflectionUtils.java:88)
    at org.apache.hadoop.util.ReflectionUtils.setConf(ReflectionUtils.java:64)
    at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:117)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:432)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:372)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
    at org.apache.hadoop.mapred.Child.main(Child.java:249)
Caused by: java.io.IOException: java.io.IOException: error=2, No such file or directory
    at java.lang.UNIXProcess.<init>(UNIXProcess.java:148)
    at java.lang.ProcessImpl.start(ProcessImpl.java:65)
    at java.lang.ProcessBuilder.start(ProcessBuilder.java:453)
    ... 24 more

可以使用Streaming的-file选项指定脚本文件加入到Job的Jar文件中，即使用上面运行的命令行中指定的“-file *.py”，而实现的2个Python脚本文件就在当前运行Job的结点当前目录下。

参考链接

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布，欢迎转载、使用、重新发布，但务必保留文章署名时延军（包含链接：http://shiyanjun.cn），不得用于商业目的，基于本文修改后的作品务必以相同的许可发布。如有任何疑问，请与我联系。

评论(5)： “Hadoop Streaming原理及实践”

鸟鹏

2015-09-24 09:52:00

为啥要使用filter？
words = filter(lambda word: word, line.split())
line.split()不就是返回列表吗？

请教楼主，谢谢！374741295@qq.com

回复
Pingback: Hadoop Streaming原理及实践 | 赛小秘
赵翔

2016-05-16 21:50:37

你好，我在hive中通过Transform用Python脚本来处理数据处理， python中引用了其他自己写的模块，都已经add file到hive中，运行时报模块找不到错误，这是为什么，难道只能写在一个python文件中吗？

回复
- feaw
  
  2016-11-02 17:12:49
  
  要写成 python file.py
  
  回复
mmm

2016-12-26 10:39:29

你好，我用的python,执行时提示，Not a valid JAR: /usr/local/hadoop/share/hadoop/tools/sources/hadoop-*streaming*.jar-file ，请问这是怎么回事，我找了，我hadoop里没有这个目录，应该怎么解决呢

回复

发表评论取消回复

gsgsgsl: 赞一个，前几年搞过kafka2.x版本的鉴权，几年有这方面需求，发现很多api变了，2.x版本的鉴权配置在3.x版...
dack: GPT-2中的相对位置编码请问有出处吗，在GPT-2的论文& #8221;Language Models are...
zhang: 你好，这一行“前面计算已经得到 QKT 矩阵，n=6，dk=8，则 A 的大小也是 6 x 6。”请问在下面的代码中为什么dk =...
derek: 何时才能出现伴侣Ai
丘比特: 请问博主，如果在窗口中用到广播状态，现在您有什么实现方案吗？
z: 寫的真好
方俊: 大佬好有耐心，从14年回复到19年哈哈
Yanjun: 图是用 Astah 和 OminiGraffle 画的
JacobZheng: 问个题外话，图是用什么工具画的啊
Derek Dekker: 感觉还挺难的
luosijie: 博主你好，请问您知道K距离方法出自哪篇文献吗，我该如何引用？
lee: 你好，点赞

评论(5)： “Hadoop Streaming原理及实践”

发表评论 取消回复

发表评论取消回复