一： Hadoop Streaming详解

1、Streaming的作用

Hadoop Streaming框架，最大的好处是，让任何语言编写的map, reduce程序能够在hadoop集群上运行；map/reduce程序只要遵循从标准输入stdin读，写出到标准输出stdout即可

其次，容易进行单机调试，通过管道前后相接的方式就可以模拟streaming, 在本地完成map/reduce程序的调试

# cat inputfile | mapper | sort | reducer > output

最后，streaming框架还提供了作业提交时的丰富参数控制，直接通过streaming参数，而不需要使用java语言修改；很多mapreduce的高阶功能，都可以通过steaming参数的调整来完成

 2、Streaming的局限

Streaming默认只能处理文本数据Textfile，对于二进制数据，比较好的方法是将二进制的key, value进行base64编码，转化为文本

Mapper和reducer的前后都要进行标准输入和标准输出的转化，涉及数据拷贝和解析，带来了一定的开销

3、Streaming命令的相关参数    (普通选项、streaming选项)

Streaming命令的形式如下：

#  /usr/local/src/hadoop-1.2.1/bin/hadoop jar  hadoop-streaming.jar \

   [普通选项]  [Streaming选项]         #  注意：普通选项一定要写在streaming选项前面

普通选项

插曲1： mapred-site.xml 指定map的slot，reduce的slot

Map和reduce在datanode上的运行，会受到slot的限制，并且有各自的slot限制； 每个Datanode读取相应的配置文件, 从而确定每个datanode上能运行的最大map,reduce个数，以及节点能力是否充分发挥

Hadoop1.0中，slot在mapred-site.xml中配置（mapreduce作业前配置好）, 基本上每个slot在运行1个map, reduce作业后会占用1个CPU core,   最激进的做法是设置map和reduce的slot都是CPU core-1 （Map执行完后才会进行reduce）,  预留1个CPU core给tasktracker（比如上报心跳等）,  但通常reducer的slot要比reducer少，考虑大多数情况下mapper要比reducer多

默认map的slot为2，reduce的slot也为2

<configuration>

        <property>

                <name>mapred.job.tracker</name>

                <value>http://192.168.179.100:9001</value>

        </property>

        <property>

               <name>mapred.tasktracker.map.tasks.maximum</name>

               <value>15</value>

        </property>

        <property>

               <name>mapreduce.tasktracker.tasks.reduce.maximum</name>

               <value>10</value>

        </property>

</configuration>

插曲二： mapred-site.xml 指定map最终输出的merge文件的存放路径

<configuration>

        <property>

                <name>mapred.job.tracker</name>

                <value>http://192.168.179.100:9001</value>

        </property>

        <property>

               <name>mapred.local.dir</name>

               <value>/usr/loca/src/hadoop-1.2.1/tmp/mapoutput</value>

        </property>

</configuration>

当1个作业被提交并在tasktracer的管理下开始运行时，会对每个job创建1个目录，所有分发的文件，都放置在这里

${mapred.local.dir}/taskTracker/$user/jobcache/$jobid/jars/

普通选项中的-D  property=value

-D    普通选项，使用最多的高级参数，替代-jobconf（参数将被废弃），需要注意的是 -D选项要放在streaming参数的前面，一般我会放在参数的开头

-D 指定job名称

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
-D mapred.job.name=”Test001”

-D 指定reduce任务、map任务个数

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \

-D mapred.job.name=”Teset001”
-D mapred.reduce.tasks=2    # reduce task个数，一定生效
-D mapred.map.tasks=5       # map task个数，不一定生效

-D 指定mapper的输出分隔符

-D stream.map.output.field.seperator=.  # 指定mapper每条输出key,value分隔符
-D stream.num.map.output.key.fields=4   # 第4个.之前的部分为key,剩余为value
-D map.output.key.field.separator=.      # 设置map输出中，Key内部的分隔符

-D 指定基于哪些key进行分桶

基于指定的Key进行分桶，打标签

指定列数

-D num.key.fields.for.partition=1       # 只用1列Key做分桶
-D num.key.fields.for.partition=2       # 使用1,2共两列key做分桶

指定某些字段做key

-D mapred.text.key.partitioner.option =-k1,2   # 第1,2列Key做分桶
-D mapred.text.key.partitioner.option =-k2,2   # 第2列key做分桶

都要修改partition为能够只基于某些Key进行分桶的类

-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner

-D 指定将reducer的输出进行压缩

-D mapred.output.compress=true

-D mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec

-D 指定将mapper的输出进行压缩

-D mapred.compress.map.output=true

-D mapred.map.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec

-D 指定Comparator对key进行数字、倒序排序

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \

-D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \  # 使用keyFieldBasedComparator进行key排序

-D stream.map.output.field.separator=. \

-D stream.num.map.output.key.fields=4 \

-D map.output.key.field.separator=. \

-D mapred.text.key.comparator.options=-k2,2nr \

# -k2,2只用第二列排序，n数字排序，r倒序(从大到小)

-input myInputDirs \

-output myOutputDir \

-mapper org.apache.hadoop.mapred.lib.IdentityMapper \

-reducer org.apache.hadoop.mapred.lib.IdentityReducer

-D 指定每个reduce task申请的内存数量

-D mapreduce.reduce.memory.mb=512  #单位为M

Streaming选项

二、Mapper输入/输出，根据哪些key分桶，根据哪些key进行排序

先看看Hadoop-1.2.1 文档原文中的解释

As the mapper task runs, it converts its inputs into lines and feed the lines to the stdin of the process. In the meantime, the mapper collects the line oriented outputs from the stdout of the process and converts each line into a key/value pair, which is collected as the output of the mapper. By default, the prefix of a line up to the first tab character is the key and the rest of the line (excluding the tab character) will be the value. If there is no tab character in the line, then entire line is considered as key and the value is null. However, this can be customized, as discussed later.

Mapper输入：

每一个mapper开始运行时，输入文件会被转换成多行(TextInputformat根据\n来进行分行)，并将每一行传递给stdin, 作为Mapper的输入， mapper直接对stdin中的每行内容做处理

Mapper输出分隔符：

默认情况下hadoop设置mapper输出的key, value通过tab进行分隔，可以重新指定

-D stream.map.output.field.seperator=.    # 指定mapper每条输出key,value分隔符

-D stream.num.map.output.key.fields=4   # 第4个.之前的部分为key,剩余为value

mapper的输出会经历：

1、 partition前，根据mapper输出分隔符分离出Key和Value；

-D stream.map.output.field.separator=.    # 指定mapper每条输出key,value分隔符

-D stream.num.map.output.key.fields=4   # 第4个.之前的为key, 剩下的为value

-D  map.output.key.field.separator=.        # 设置map输出中，Key内部的分隔符

2、 根据 “分桶分隔符”，确定哪些key被用来做partition（默认是用所有key, 只有1列; 或者是Mapper输出分隔符分离出的所有key都被用于Partition）

基于指定的Key进行分桶，打标签

指定列数

-D num.key.fields.for.partition=1        # 只用1列Key做分桶，也就是第一列

-D num.key.fields.for.partition=2       # 使用1,2共两列key做分桶（列数）

指定某些字段做key

-D mapred.text.key.partitioner.option =-k1,2   # 第1,2列Key做分桶

-D mapred.text.key.partitioner.option =-k2,2   # 第2列key做分桶

#都要修改partition为能够只基于某些Key进行分桶的类

-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner

3、Spill时根据Partition标签和所有Key进行排序；

4、Partition标签和key之间，也是通过mapper输出分隔符来隔离

5、reducer前的文件会删除partition标签，并根据Mapper输出分隔符确定出key, 进行Reducer前的归并排序；（reducer前的归并排序，基于所有mapper的key进行排序）

因此如果要定义新的Mapper输出分隔符就要做到：1）mapper代码中根据新分隔符来构建输出到stdout的内容；2）提交作业时，通过—D 指定新的Mapper输出分隔符，以及第几个分隔符来分离Key

Reducer的输入：

每个Reducer的每条输入，就是去除Partition标签（根据Mapper分隔符分离出partition标签）后的内容，和Mapper输出到stdout中的内容相同，但不同记录之间已经做了排序；因此如果重新指定了Mapper的输出分隔符，Reducer程序就要修改为根据新的Mapper输出分隔符来分离Key,value;

Reducer的输出：

Reducer的输出，默认也是根据tab来分离key,value, 这也是reducer程序要根据tab来组合key,value输出给stdout的原因；  当Reducer输出分隔符重新指定，Reducer程序中输出给stdout的内容也要配合新的分隔符来构造（Reducer->stdout-> outputformat ->file,  outputformat根据reducer的输出分隔符来分离key,value,  并写入文件）

-D stream.reduce.output.field.seperator=.      # reducer输出key,value间的分隔符

-D stream.num.reduce.output.key.fields=4     # 第4个.之前的内容为key, 其他为value