一、Hadoop Streaming
它是hadoop的一个工具,用来创建和运行一类特殊的map/reduce作业。所谓的特殊的map/reduce作业可以是可执行文件或脚本本件(Python、PHP、c等)。Streaming使用“标准输入”和“标准输出”与我们编写的Map和Reduce进行数据的交换。由此可知,任何能够使用“标准输入”和“标准输出”的编程语言都可以用来编写MapReduce程序。如下
$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-streaming.jar
-input myInputDirs 【hdfs系统中输入文件/夹位置】
-output myOutputDir
-mapper cat
-reducer wc
-file 【需要分发的文件】
二、Hadoop Streaming原理
在上面的例子里,mapper和reducer都是可执行文件,它们从标准输入读入数据(一行一行读),并把计算结果发给标准输出。Streaming工具会创建一个Map/Reduce作业,并把它发送给合适的集群,同时监视这个作业的整个执行过程。
如果一个可执行文件被用于mapper,则在mapper初始化时,每一个mapper任务会把这个可执行文件作为一个单独的进程启动。 mapper任务运行时,它把输入切分成行并把每一行提供给可执行文件进程的标准输入。同时,mapper收集可执行文件进程标准输出的内容,并把收到的每一行内容转化成key/value对,作为mapper的输出。默认情况下,一行中第一个tab之前的部分作为key,之后的(不包括tab)作为value。如果没有tab,整行作为key值,value值为null。不过,这可以定制,在下文中将会讨论如何自定义key和value的切分方式。
如果一个可执行文件被用于reducer,每个reducer任务会把这个可执行文件作为一个单独的进程启动。 Reducer任务运行时,它把输入切分成行并把每一行提供给可执行文件进程的标准输入。同时,reducer收集可执行文件进程标准输出的内容,并把每一行内容转化成key/value对,作为reducer的输出。默认情况下,一行中第一个tab之前的部分作为key,之后的(不包括tab)作为value。在下文中将会讨论如何自定义key和value的切分方式。
三、使用注意事项
1、使用Hadoop Streaming运行MapReduce会比用java代码写的MapReduce稍慢一些,原因如下:
Java运行Map任务输出一定数量结果集后就会启动Reduce任务,而使用Streaming要等到所有Map任务都执行完毕后才会启动Reduce任务;
2、运行失败
使用Hadoop Streaming时需要将mapper文件和reducer文件放到所有tasktracker节点上。或者采用 -file 选型指定文件,打包文件到提交的作业中,可以是mapper或者reducer要用的输入文件,如配置文件,字典,framework会把文件当成作业提交的一部分分发到集群的机器上。。
3、只需要map函数处理数据,如何设置?
“-jobconf mapred.reduce.tasks=0” ,Map/Reduce框架就不会创建reducer任务,直接使用mapper任务的输出做为最终输出。
4、如何为作业指定其他插件
-inputformat JavaClassName
-outputformat JavaClassName
-partitioner JavaClassName 【用户自定义的partitioner程序】
-combiner JavaClassName 【用户自定义的combiner程序(必须用java实现)】
5、为作业指定附加配置参数
【-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所占的域数目 6、如何处理python程序map输出文件,每行后面的的tab符 or 空格符; 问题原因:当Map/Reduce框架从mapper的标准输入读取一行时,它把这一行切分为key/value对。在默认情况下,每行第一个tab符之前的部分作为key,之后的部分作为value(不包括tab符)。 参考5做如下设置:建设分隔符是"^"【-jobconf mapred.textoutputformat.separator=^ -jobconf stream.map.output.field.separator=^】 7、如何输出gzip文件格式的输出 你只需设置streaming作业中的选项‘-jobconf mapred.output.compress=true -jobconf mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCode’ 8、如何使用Streaming如何解析XML文档? 可以使用StreamXmlRecordReader来解析XML文档。hadoop jar hadoop-streaming.jar -inputreader "StreamXmlRecord,begin=BEGIN_STRING,end=END_STRING" ..... (rest of the command) Map任务会把BEGIN_STRING和END_STRING之间的部分看作一条记录。