课程表

Hadoop课程

工具箱
速查手册

Hadoop 流

当前位置:免费教程 » 大数据/云 » Hadoop

Hadoop流是Hadoop发行版附带的一个实用程序。此实用程序允许您使用任何可执行文件或脚本作为映射程序和/或reducer创建和运行Map / Reduce作业。

原理

Hadoop Streaming是Hadoop提供的一个编程工具,它允许用户使用任何可执行文件或者脚本文件作为Mapper和Reducer,例如:采用shell脚本语言中的一些命令作为mapper和reducer(cat作为mapper,wc作为reducer)

  1. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar -input myInputDirs -output myOutputDir -mapper cat -reducer wc

mapper和reducer会从标准输入中读取用户数据,一行一行处理后发送给标准输出。Streaming工具会创建MapReduce作业,发送给各个tasktracker,同时监控整个作业的执行过程。

如果一个文件(可执行或者脚本)作为mapper,则在mapper初始化时,每一个mapper任务会把该文件作为一个单独进程启动,mapper任务运行时,它把输入切分成行并把每一行提供给可执行文件进程的标准输入。 同时,mapper收集可执行文件进程标准输出的内容,并把收到的每一行内容转化成key/value对,作为mapper的输出。默认情况下,一行中第一个tab之前的部分作为key,之后的(不包括tab)作为value。如果没有tab,整行作为key值,value值为null。不过,这可以定制,在下文中会介绍如何自定义key和value的切分方式。

对于reducer,类似。

以上是Map/Reduce框架和streaming mapper/reducer之间的基本通信协议。

语法

基本语法

  1. 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所占的域数目。

有时只需要map函数处理输入数据,这时只需把mapred.reduce.tasks设置为零,Map/Reduce框架就不会创建reducer任务,mapper任务的输出就是整个作业的最终输出。为了做到向下兼容,Hadoop Streaming也支持“-reduce None”选项,它与“-jobconf mapred.reduce.tasks=0”等价。

扩展语法

之前已经提到,当Map/Reduce框架从mapper的标准输入读取一行时,它把这一行切分为key/value对。在默认情况下,每行第一个tab符之前的部分作为key,之后的部分作为value(不包括tab符)。

但是,用户也可以自定义,可以指定分隔符是其它字符而不是默认的tab符,或者指定在第n(n>=1)个分割符处分割而不是默认的第一个。例如:

  1. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar -input myInputDirs -output myOutputDir -mapper org.apache.hadoop.mapred.lib.IdentityMapper -reducer org.apache.hadoop.mapred.lib.IdentityReducer -jobconf stream.map.output.field.separator=. -jobconf stream.num.map.output.key.fields=4

在上面的例子中,“-jobconf stream.map.output.field.separator=.”指定“.”作为map输出内容的分隔符,并且从在第4个“.”之前的部分作为key,之后的部分作为value(不包括这第4个“.”)。 如果一行中的“.”少于4个,则整行的内容作为key,value设为空的Text对象(就像这样创建了一个Text:new Text(""))。

同样地,用户也可以使用“-jobconf stream.reduce.output.field.separator=SEP”和“-jobconf stream.num.reduce.output.fields=NUM”来指定reduce输出的行中,第几个分隔符处分割key和value。


实例

为了说明各种语言编写Hadoop Streaming程序的方法,下面以WordCount为例,WordCount作业的主要功能是对用户输入的数据中所有字符串进行计数。

1、shell

#vi mapper.sh

  1. #! /bin/bash
  2. while read LINE; do
  3. for word in $LINE
  4. do
  5. echo "$word 1"
  6. done
  7. done

#vi reducer.sh

  1. #! /bin/bash
  2. count=0
  3. started=0
  4. word=""
  5. while read LINE;do
  6. newword=`echo $LINE | cut -d ' ' -f 1`
  7. if [ "$word" != "$newword" ];then
  8. [ $started -ne 0 ] && echo -e "$word\t$count"
  9. word=$newword
  10. count=1
  11. started=1
  12. else
  13. count=$(( $count + 1 ))
  14. fi
  15. done
  16. echo -e "$word\t$count"

本地测试:

  1. cat input.txt | sh mapper.sh | sort | sh reducer.sh

集群测试:

  1. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar -input myInputDirs -output myOutputDir -mapper mapper.sh -reducer reducer.sh

如果执行上面脚本提示:“Caused by: java.io.IOException: Cannot run program “/user/hadoop/Mapper”: error=2, No such file or directory”,则说明找不到可执行程序,可以在提交作业时,采用-file选项指定这些文件,比如上面例子中,可以使用“-file mapper.py -file reducer.py”,这样,Hadoop会将这两个文件自动分发到各个节点上,比如:

  1. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar -input myInputDirs -output myOutputDir -mapper mapper.sh -reducer reducer.sh -file mapper.sh -file reducer.sh

2、python

#vi mapper.py

  1. #!/usr/bin/env python
  2. import sys
  3. #maps words to their counts
  4. word2count = {}
  5. #input comes from STDIN (standard input)
  6. for line in sys.stdin:
  7. #remove leading and trailing whitespace
  8. line = line.strip()
  9. #split the line into words while removing any empty strings
  10. words = filter(lambda word: word, line.split())
  11. #increase counters
  12. for word in words:
  13. #write the results to STDOUT (standard output);
  14. #what we output here will be the input for the
  15. #Reduce step, i.e. the input for reducer.py
  16. #
  17. #tab-delimited; the trivial word count is 1
  18. print '%s\t%s' % (word, 1)

#vi reducer.py

  1. #!/usr/bin/env python
  2. from operator import itemgetter
  3. import sys
  4. #maps words to their counts
  5. word2count = {}
  6. #input comes from STDIN
  7. for line in sys.stdin:
  8. #remove leading and trailing whitespace
  9. line = line.strip()
  10. #parse the input we got from mapper.py
  11. word, count = line.split()
  12. #convert count (currently a string) to int
  13. try:
  14. count = int(count)
  15. word2count[word] = word2count.get(word, 0) + count
  16. except ValueError:
  17. #count was not a number, so silently
  18. #ignore/discard this line
  19. pass
  20. #sort the words lexigraphically;
  21. #
  22. #this step is NOT required, we just do it so that our
  23. #final output will look more like the official Hadoop
  24. #word count examples
  25. sorted_word2count = sorted(word2count.items(), key=itemgetter(0))
  26. #write the results to STDOUT (standard output)
  27. for word, count in sorted_word2count:
  28. print '%s\t%s'% (word, count)

本地测试:

  1. cat input.txt | python mapper.py | sort | python reducer.py

集群测试:

  1. $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-*-streaming.jar -input myInputDirs -output myOutputDir -mapper mapper.py -reducer reducer.py

使用Python的示例

对于Hadoop流,我们正在考虑字数问题。 Hadoop中的任何作业必须有两个阶段:mapper和reducer。我们已经为python脚本中的mapper和reducer编写了代码,以便在Hadoop下运行它。也可以在Perl和Ruby中写同样的内容。

映射器阶段代码

  1. !/usr/bin/python
  2. import sys
  3. # Input takes from standard input for myline in sys.stdin:
  4. # Remove whitespace either side myline = myline.strip()
  5. # Break the line into words words = myline.split()
  6. # Iterate the words list for myword in words:
  7. # Write the results to standard output print '%s %s' % (myword, 1)

确保此文件具有执行权限(chmod + x /home/expert /hadoop-1.2.1 / mapper.py)。

减速器阶段代码

  1. #!/usr/bin/python
  2. from operator import itemgetter
  3. import sys
  4. current_word = ""
  5. current_count = 0
  6. word = ""
  7. # Input takes from standard input for myline in sys.stdin:
  8. # Remove whitespace either side myline = myline.strip()
  9. # Split the input we got from mapper.py word, count = myline.split(' ', 1)
  10. # Convert count variable to integer
  11. try:
  12. count = int(count)
  13. except ValueError:
  14. # Count was not a number, so silently ignore this line continue
  15. if current_word == word:
  16. current_count += count
  17. else:
  18. if current_word:
  19. # Write result to standard output print '%s %s' % (current_word, current_count)
  20. current_count = count
  21. current_word = word
  22. # Do not forget to output the last word if needed!
  23. if current_word == word:
  24. print '%s %s' % (current_word, current_count)

将mapper和reducer代码保存在Hadoop主目录中的mapper.py和reducer.py中。确保这些文件具有执行权限(chmod + x mapper.py和chmod + x reducer.py)。因为python是缩进敏感所以相同的代码可以从下面的链接下载。

执行WordCount程序

  1. $ $HADOOP_HOME/bin/hadoop jar contrib/streaming/hadoop-streaming-1.
  2. 2.1.jar
  3. -input input_dirs
  4. -output output_dir
  5. -mapper <path/mapper.py
  6. -reducer <path/reducer.py

其中“\”用于行连续以便清楚可读性。

例如:

  1. ./bin/hadoop jar contrib/streaming/hadoop-streaming-1.2.1.jar -input myinput -output myoutput -mapper /home/expert/hadoop-1.2.1/mapper.py -reducer /home/expert/hadoop-1.2.1/reducer.py

流如何工作

在上面的示例中,mapper和reducer都是从标准输入读取输入并将输出发送到标准输出的python脚本。该实用程序将创建一个Map / Reduce作业,将作业提交到适当的群集,并监视作业的进度,直到作业完成。

当为映射器指定脚本时,每个映射器任务将在映射器初始化时作为单独的进程启动脚本。当映射器任务运行时,它将其输入转换为行,并将这些行馈送到进程的标准输入(STDIN)。同时,映射器从进程的标准输出(STDOUT)收集面向行的输出,并将每行转换为键/值对,作为映射器的输出收集。默认情况下,直到第一个制表符字符的行的前缀是键,行的其余部分(不包括制表符字符)将是值。如果行中没有制表符,则整个行被视为键,值为null。但是,这可以根据一个需要定制。

当为reducer指定脚本时,每个reducer任务将作为单独的进程启动脚本,然后初始化reducer。当reducer任务运行时,它将其输入键/值对转换为行,并将行馈送到进程的标准输入(STDIN)。同时,reducer从进程的标准输出(STDOUT)收集面向行的输出,将每行转换为键/值对,将其作为reducer的输出进行收集。默认情况下,直到第一个制表符字符的行的前缀是键,行的其余部分(不包括制表符字符)是值。但是,这可以根据特定要求进行定制。

重要命令

参数描述
-input directory/file-name输入mapper的位置。(需要)
-output directory-name减速器的输出位置。(需要)
-mapper executable or script or JavaClassNameMapper可执行文件。(需要)
-reducer executable or script or JavaClassNameReducer可执行文件。(需要)
-file file-name使mapper,reducer或combiner可执行文件在计算节点本地可用。
-inputformat JavaClassName你提供的类应该返回Text类的键/值对。如果未指定,则使用TextInputFormat作为默认值。
-outputformat JavaClassName您提供的类应该采用Text类的键/值对。如果未指定,则使用TextOutputformat作为默认值。
-partitioner JavaClassName确定将键发送到哪个reduce的类。
-combiner streamingCommand or JavaClassName组合器可执行映射输出。
-cmdenv name=value将环境变量传递到流式命令。
-inputreader对于向后兼容性:指定记录读取器类(而不是输入格式类)。
-verbose详细输出。
-lazyOutput创建输出延迟。例如,如果输出格式基于FileOutputFormat,则输出文件仅在首次调用output.collect(或Context.write)时创建。
-numReduceTasks指定Reducer的数量。
-mapdebug映射任务失败时调用的脚本。
-reducedebug当reduce任务失败时调用的脚本。
本节参考:http://bbs.csdn.net/topics/390909413
转载本站内容时,请务必注明来自W3xue,违者必究。
 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号